Redis 任務隊列(生產者消費者)

本文轉載自查看原文 2017-10-10 14:00 1862 java

安裝reids參考下面網址：

http://www.runoob.com/redis/redis-install.html

pom.xml中的jar包

 1 <dependency>
 2         <groupId>org.springframework.data</groupId>
 3         <artifactId>spring-data-redis</artifactId>
 4         <version>1.6.0.RELEASE</version>
 5     </dependency>
 6 
 7     <!-- jedis -->
 8     <dependency>
 9           <groupId>redis.clients</groupId>
10           <artifactId>jedis</artifactId>
11            <version>2.5.1</version>
12     </dependency>

Jedis的工具類JedisUtil

  1 package demo3;
  2 
  3 import java.util.List;
  4 
  5 import redis.clients.jedis.Jedis;
  6 public class JedisUtil {
  7     private static Jedis jedis = null;
  8     /**
  9      * 存儲REDIS隊列 順序存儲
 10      * @param key 字節類型
 11      * @param value 字節類型
 12      */
 13     public static void lpush(byte[] key,byte[] value){
 14         try {
 15             jedis = RedisPool.getJedis();
 16             jedis.lpush(key,value);
 17         } catch (Exception e) {
 18             e.printStackTrace();
 19         }finally {
 20             RedisPool.returnResource(jedis);
 21         }
 22     }
 23     /**
 24      * 存儲REDIS隊列 反序存儲
 25      * @param key 字節類型
 26      * @param value 字節類型
 27      */
 28     public static void rpush(byte[] key,byte[] value){
 29         try {
 30             jedis = RedisPool.getJedis();
 31             jedis.rpush(key,value);
 32         } catch (Exception e) {
 33             e.printStackTrace();
 34         }finally {
 35             RedisPool.returnResource(jedis);
 36         }
 37     }
 38     /**
 39      * 移除列表的最后一個元素，並將該元素添加到另一個列表並返回,就可以實現任務隊列
 40      * @param srckey 原隊列的key
 41      * @param dstkey 目標隊列的key
 42      */
 43     public static byte[] rpoplpush(byte[] srckey,byte[] dstkey){
 44         byte[] value = null;
 45         try {
 46             jedis = RedisPool.getJedis();
 47             value= jedis.rpoplpush(srckey,dstkey);
 48         } catch (Exception e) {
 49             e.printStackTrace();
 50         }finally {
 51             RedisPool.returnResource(jedis);
 52         }
 53         return value;
 54     }
 55     /**
 56      * 從列表中彈出一個值，將彈出的元素插入到另外一個列表中並返回它； 如果列表沒有元素會阻塞列表直到等待超時或發現可彈出元素為止。
 57      * @param srckey
 58      * @param dstkey
 59      * @param timout
 60      * @return
 61      */
 62     public static byte[] brpoplpush(byte[] srckey,byte[] dstkey,int timout){
 63         byte[] value = null;
 64         try {
 65             jedis = RedisPool.getJedis();
 66             value = jedis.brpoplpush(srckey,dstkey,timout);
 67         } catch (Exception e) {
 68             e.printStackTrace();
 69         } finally {
 70             RedisPool.returnResource(jedis);
 71         }
 72         return value;
 73     }
 74     /**
 75      * 設置實現任務隊列的鍵和過期時間
 76      * @param key
 77      * @param timeout
 78      */
 79     public static List<byte[]> brpop(byte[] key, int timeout){
 80         List<byte[]> result = null;
 81         try {
 82             jedis = RedisPool.getJedis();
 83             result=jedis.brpop(0,key);
 84         } catch (Exception e) {
 85             e.printStackTrace();
 86         } finally {
 87             RedisPool.returnResource(jedis);
 88         }
 89         return result;
 90     }
 91     /**
 92      * 移除隊列中的最后一個元素並顯示最后一個元素
 93      * @param key
 94      * @return
 95      */
 96     public static byte[] rpop(byte[] key) {
 97         byte[] bytes = null;
 98         try {
 99             jedis = RedisPool.getJedis();
100             bytes = jedis.rpop(key);
101         } catch (Exception e) {
102             e.printStackTrace();
103         } finally {
104             RedisPool.returnResource(jedis);
105         }
106         return bytes;
107     }
108 }

用到的工具類RedisPool

 1 package demo3;
 2 
 3 import redis.clients.jedis.Jedis;
 4 import redis.clients.jedis.JedisPool;
 5 import redis.clients.jedis.JedisPoolConfig;
 6 
 7 public class RedisPool {
 8      private static final JedisPool jedisPool;
 9      private static final String REDIS_IP="127.0.0.1";
10      private static final int REIDSPORT=6379;
11      
12     static {
13         jedisPool=new JedisPool(new JedisPoolConfig(), REDIS_IP, REIDSPORT);
14     }
15     
16     public static Jedis getJedis(){
17         return jedisPool.getResource();
18     }
19     
20     public static void returnResource(Jedis jedis){
21         jedisPool.returnResource(jedis);
22     }
23 }

用到的工具類SerialoizebleUtil

package demo3;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;

public class SerialoizebleUtil {
public static Object ByteToObject(byte[] bytes) {  
 Object obj = null;  
 try {  
     // bytearray to object  
     ByteArrayInputStream bi = new ByteArrayInputStream(bytes);  
     ObjectInputStream oi = new ObjectInputStream(bi);  
   
     obj = oi.readObject();  
     bi.close();  
     oi.close();  
 } catch (Exception e) {  
     System.out.println("translation" + e.getMessage());  
     e.printStackTrace();  
 }  
        return obj;  
 }

public static byte[] ObjectToByte(java.lang.Object obj) {  
    byte[] bytes = null;  
    try {  
        // object to bytearray  
        ByteArrayOutputStream bo = new ByteArrayOutputStream();  
        ObjectOutputStream oo = new ObjectOutputStream(bo);  
        oo.writeObject(obj);  
  
        bytes = bo.toByteArray();  
  
        bo.close();  
        oo.close();  
    } catch (Exception e) {  
        System.out.println("translation" + e.getMessage());  
        e.printStackTrace();  
    }  
    return bytes;  
}
}

實體類MessageUtil，實現存入Redis中的是對象，不是單單的基本類型，存入Redis中的對象需要實現序列化接口

 1 package model;
 2 
 3 import java.io.Serializable;
 4 
 5 public class Message implements Serializable{
 6 
 7     /**
 8      * 
 9      */
10     private static final long serialVersionUID = 1L;
11 
12     private String titile;
13     private String info;
14      
15     public Message(String titile,String info){
16     this.titile=titile;
17     this.info=info;
18     }
19 
20     public String getTitile() {
21         return titile;
22     }
23 
24     public void setTitile(String titile) {
25         this.titile = titile;
26     }
27 
28     public String getInfo() {
29         return info;
30     }
31 
32     public void setInfo(String info) {
33         this.info = info;
34     }
35     
36     
37 }

使用兩個Redis列表，一個隊列作為生成者，一個隊列作為消費者，加上線程實現兩個列表，一個列表產生任務，通過任務隊列，另一個列表處理任務

 1 package demo3;
 2 
 3 import redis.clients.jedis.Jedis;
 4 
 5 public class InitList {
 6     public static byte[] rediskey = "key".getBytes();
 7     public static byte[] dstkey = "dstkey".getBytes();
 8     public static long time = 0;
 9     public static int i = 0;
10 
11     public static void main(String args[]) {
12         Jedis jedis = RedisPool.getJedis();
13         while (true) {
14             try {
15                 MessageUtil msg1 = new MessageUtil();
16                 msg1.setId(i);
17                 msg1.setContent("wq" + i);
18                 // JedisUtil.lpush(rediskey,SerialoizebleUtil.serialize(msg1));
19                 JedisUtil.lpush(rediskey, SerialoizebleUtil.ObjectToByte(msg1));
20                 time = 2000;
21                 System.out.println("success" + i);
22                 System.out.println(jedis.lrange(rediskey, 0, 100));
23                 i++;
24                 Thread.sleep(time);
25             } catch (InterruptedException e) {
26                 e.printStackTrace();
27             }
28         }
29     }
30 }

控制台輸出如下，說明一直在向列表中插入新產生的對象

再開啟一個線程，使用Redis中的brpoplpush方法，實現任務隊列原理

 1 package demo3;
 2 
 3 import redis.clients.jedis.Jedis;
 4 
 5 public class JedisUtilTest {
 6     public static byte[] rediskey = "key".getBytes();
 7     public static byte[] dstkey = "dstkey".getBytes();
 8     public static long time = 0;
 9 
10     public static void main(String args[]) {
11         Jedis jedis = RedisPool.getJedis();
12         while (true) {
13             try {
14                 byte[] bytes = JedisUtil.brpoplpush(rediskey, dstkey, 0);
15                 MessageUtil msg = (MessageUtil) SerialoizebleUtil.ByteToObject(bytes);
16                 if (msg != null) {
17                     System.out.println(msg.getId() + " " + msg.getContent());
18                 }
19                 time = 3000;
20                 System.out.println(jedis.lrange(rediskey, 0, 100));
21                 System.out.println(jedis.lrange(dstkey, 0, 100));
22                 Thread.sleep(time);
23             } catch (InterruptedException e) {
24                 e.printStackTrace();
25             }
26         }
27     }
28 }

控制台輸出如下，上面的列表存的是剛才產生的6個對象，下面圈出來的是新的列表，可以看出新的列表的對象在遞增，說明成功實現了任務隊列原理

免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。

猜您在找 Redis實現消息隊列（生產者/消費者、發布訂閱模式）阻塞隊列實現生產者消費者模式雙緩沖隊列,生產者消費者模式並發無鎖隊列學習之二【單生產者單消費者】 disruptor 單生產者多消費者生產者和消費者模型生產者與消費者問題 RocketMQ的生產者和消費者生產者和消費者模型 Dubbo生產者和消費者