Redis（四）——消息队列

Redis不仅可作为缓存服务器，还可用作消息队列。它的列表类型天生支持用作消息队列。

**性质：**由于Redis的列表是使用双向链表实现的，保存了头尾节点，所以在列表头尾两边插取元素都是非常快的。

所以可以直接使用Redis的List实现消息队列，只需简单的两个指令lpush和rpop或者rpush和lpop。

（列表常用命令）

RPUSH : RPUSH key-name value [value1 value2,...] ------------将一个或多个值推入列表右端

LPUSH ： LPUSH key-name value [value1 value2,...] ------------将一个或多个值推入列表左端

RPOP ： RPOP key-name----------移除并返回列表最右端元素

LPOP ：LPOP key-name----------移除并返回列表最左端元素

LINDEX ： LINDEX key-name offset --------------返回列表中偏移量为offset的元素

LRANGE : LRANGE key-name start end -------------返回列表中偏移量从start到end范围内的元素

LTRIM : LTRIM key-name start end ----------------对列表进行修剪，只保留偏移量从start到end范围内的元素

其中简单示例如下：首先连接redis服务器，其中我应用了Jedispool，代码如下：

1package redis;
2
3import java.io.IOException;
4import java.util.Properties;
5
6import org.springframework.core.io.support.PropertiesLoaderUtils;
7
8import redis.clients.jedis.Jedis;
9import redis.clients.jedis.JedisPool;
10import redis.clients.jedis.JedisPoolConfig;
11/**
12 * redis单例连接池
13 * @author admin
14 *
15 */
16public class RedisPool {
17
18    private static  int TIMEOUT = 1000*30;
19    private static  int MAXTOTAL = 1024;
20    private static  int MAXIDLE = 100;
21    private static  String REDISIP = "bei1";
22    private static  int PORT = 6379;
23    private static  String PASSWORD ="default";
24
25    static {
26        try {
27            Properties prop = PropertiesLoaderUtils.loadAllProperties("redis.properties");
28            TIMEOUT = Integer.parseInt(prop.getProperty("TIMEOUT","300000"));
29            MAXTOTAL = Integer.parseInt(prop.getProperty("MAXTOTAL","1024"));
30            MAXIDLE = Integer.parseInt(prop.getProperty("MAXIDLE","100"));
31            REDISIP = prop.getProperty("REDISIP","127.0.0.1");
32            PORT = Integer.parseInt(prop.getProperty("PORT","6379"));
33            PASSWORD = prop.getProperty("PASSWORD","default");
34        } catch (IOException e) {
35            // TODO Auto-generated catch block
36            e.printStackTrace();
37        }
38    }
39
40    private static JedisPool[] pool = new JedisPool[10];
41
42    private  RedisPool() {}
43
44    private static JedisPool getPool(int database) {
45        if(database>10) {
46            return null;
47        }
48        if(pool[database] == null) {
49            JedisPoolConfig config = new JedisPoolConfig();
50            config.setMaxTotal(MAXTOTAL);
51            config.setMaxIdle(MAXIDLE);
52            config.setMaxWaitMillis(TIMEOUT);
53            config.setTestOnBorrow(true);
54            pool[database] = new JedisPool(config,REDISIP,PORT,TIMEOUT,PASSWORD,database);
55        }
56        return pool[database];
57    }
58    //单例获取redis连接资源
59    public static Jedis getResource(int database) {
60        if(database>10) {
61            return null;
62        }
63        Jedis jedis = null;
64        if(pool[database] == null) {
65            synchronized(RedisPool.class) {
66                try {
67                    if(pool[database] == null) {
68                        pool[database] = getPool(database);
69                        try {
70                            if (pool[database] != null) {
71                                jedis = pool[database].getResource();
72                            }
73                        } catch (Exception e) {
74                            e.printStackTrace();
75                        }
76                    }
77                }catch (Exception e) {
78                    e.printStackTrace();
79                }
80            }
81        }else {
82            jedis = pool[database].getResource();
83        }
84        return jedis;
85    }
86
87}
88

定义一个生产者，代码：

1package RedisMq;
2
3import com.sun.deploy.util.StringUtils;
4import redis.RedisPool;
5import redis.clients.jedis.Jedis;
6
7import java.util.concurrent.TimeUnit;
8
9/**
10 * <p>  </p>
11 *
12 * @author ly
13 * @since 2019/1/5
14 */
15public class Producer extends Thread{
16
17    public static final String MESSAGE_KEY = "queue";
18    private Jedis jedis;
19    private String produceName;
20    private volatile int count;
21
22    public Producer(String name){
23        this.produceName = name;
24        init();
25    }
26    private void init(){
27        jedis = RedisPool.getResource(1);
28
29    }
30    public void putMessage(String message) {
31        Long size = jedis.lpush(MESSAGE_KEY, message);
32        System.out.println(produceName + ": 当前未被处理消息条数为:" + size);
33        count++;
34    }
35
36    public int getCount() {
37        return count;
38    }
39    @Override
40    public void run() {
41        try {
42            while (true) {
43                putMessage("hello world");
44                TimeUnit.SECONDS.sleep(1);
45            }
46        } catch (InterruptedException e) {
47
48        } catch (Exception e) {
49            e.printStackTrace();
50        }
51    }
52
53    public static void main(String[] args) throws InterruptedException {
54        Producer producer = new Producer("myProducer");
55        producer.start();
56
57        for (; ; ) {
58            System.out.println("main : 已存储消息条数:" + producer.getCount());
59            TimeUnit.SECONDS.sleep(10);
60        }
61    }
62}

再定义一个消费者

1package RedisMq;
2
3import redis.RedisPool;
4import redis.clients.jedis.Jedis;
5/**
6 * <p>  </p>
7 *
8 * @author ly
9 * @since 2019/1/7
10 */
11
12
13    /**
14     * 消息消费者
15     * @author yamikaze
16     */
17    public class Customer extends Thread{
18
19        private String customerName;
20        private volatile int count;
21        private Jedis jedis;
22
23        public Customer(String name) {
24            this.customerName = name;
25            init();
26        }
27
28        private void init() {
29            jedis = RedisPool.getResource(1);
30        }
31
32        public void processMessage() {
33            String message = jedis.rpop(Producer.MESSAGE_KEY);
34            if(message != null) {
35                count++;
36                handle(message);
37            }
38        }
39
40        public void handle(String message) {
41            System.out.println(customerName + " 正在处理消息,消息内容是: " + message + " 这是第" + count + "条");
42        }
43
44        @Override
45        public void run() {
46            while (true) {
47                processMessage();
48            }
49        }
50
51        public static void main(String[] args) {
52            Customer customer = new Customer("小花");
53            customer.start();
54        }
55    }
56

运行后生产者和消费者控制台信息分别如下：

Redis 发布与订阅

redis 支持消息队列。发布订阅即是一种消息通信模式：发送者发送消息，订阅者订阅消息。

redis 客户端可以订阅任意数量的频道

（一）发布订阅 使用 publish 指令，格式为 publish channel message

1127.0.0.1:6379> publish fruit "apple"
2(integer) 0
3

该返回值为0，说明没有人订阅

（二）订阅消息 使用subscribe指令接受消息，格式为 subscribe channel

1127.0.0.1:6379> subscribe fruit
2Reading messages... (press Ctrl-C to quit)
31) "subscribe"
42) "fruit"
53) (integer) 1
6

可以看到使用SUBSCRIBE指令后进入了订阅模式，但没有接收到publish发送的消息，这是因为只有在消息发出去前订阅才会接收到。在这个模式下其他指令，只能看到回复。回复信息分为3类： 1 如果为subscribe，第二个值表示订阅的频道，如上述代码

2 如果为message(消息)，第二个值为产生该消息的频道，第三个值为消息，如图：

3 如果退订消息，第二个值表示取消订阅的频道，第三个值表示当前客户端的订阅数量。则接受信息如下

（三）取消订阅 使用Unsubscribe 指令，格式为 UNSUBSCRIBE channel [channel ...]

1127.0.0.1:6379>  unsubscribe fruit
21) "unsubscribe"
32) "fruit"
43) (integer) 0

参考文章https://blog.csdn.net/qq_34212276/article/details/78455004

Redis（四）——消息队列

Redis 发布与订阅

评论区

相关推荐