使用Redis集合List实现消息队列

Redis是一个key-value存储系统。和Memcached类似，它支持存储的value类型相对更多，包括string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash（哈希类型）。这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作，而且这些操作都是原子性的。在此基础上，redis支持各种不同方式的排序。与memcached一样，为了保证效率，数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。

之前，使用Redis的List做过一个浏览记录的功能 https://www.javacui.com/DB/679.html 。

在看到使用Redis做消息队列时，有人使用List来做队列，研究了一下，确实可以实现功能。它主要是通过 lpush、rpop 存入和读取实现消息队列的。

lpush 可以把最新的消息存储到消息队列（List 集合）的首部，而 rpop 可以读取消息队列的尾部，这样就实现了先进先出。

使用 List 实现消息队列的优点是消息可以被持久化，List 可以借助 Redis 本身的持久化功能，AOF 或者是 RDB 或混合持久化的方式，用于把数据保存至磁盘，这样当 Redis 重启之后，消息不会丢失。
但使用 List 同样存在一定的问题，比如消息不支持重复消费、没有按照主题订阅的功能、不支持消费消息确认等。

使用命令发布数据：

lpush List:Javacui 111
lpush List:Javacui 222

查看数据

消费数据

127.0.0.1:0>rpop List:Javacui
"111"
127.0.0.1:0>rpop List:Javacui
"222"
127.0.0.1:0>

编码实现，使用Jedis编码

package com.example.springboot;
import redis.clients.jedis.Jedis;
public class Subscribe2 {
    public static void main(String[] args) throws InterruptedException {
        // 启动一个线程作为消费者
        new Thread(() -> bConsumer()).start();
        // 生产者
        producer();
    }
    /**
     * 生产者
     */
    public static void producer() throws InterruptedException {
        Jedis jedis = new Jedis("127.0.0.1",6379);
        jedis.auth("l52u27lv1Jur");
        // 推送消息
        // 推送消息
        jedis.lpush("javacui", "java小强 1.", "java小强 11.","java小强 1111.");
        Thread.sleep(1000);
        jedis.lpush("javacui", "java小强 2.");
    }
    /**
     * 消费者
     */
    public static void consumer() {
        Jedis jedis = new Jedis("127.0.0.1",6379);
        jedis.auth("l52u27lv1Jur");
        while (true) {
            // 以原子方式返回并删除列表的第一个 （LPOP） 或最后一个 （RPOP） 元素。例如，如果列表包含元素“a”，“b”，“c”，则RPOP将返回“c”，列表将变为“a”，“b”。
            // 如果键不存在或列表已为空，则返回特殊值“nil”。
            String msg = jedis.rpop("javacui");
            if (msg != null) {
                // 接收到了消息
                System.out.println("接收到消息：" + msg);
            }
        }
    }
    /**
     * 消费者（阻塞版）
     * brpop 中的 b 是 blocking 的意思，表示阻塞读，也就是当队列没有数据时，它会进入休眠状态
     * 当有数据进入队列之后，它才会“苏醒”过来执行读取任务，这样就可以解决 while 循环一直执行消耗系统资源的问题
     */
    public static void bConsumer() {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        jedis.auth("l52u27lv1Jur");
        while (true) {
            // BLPOP（和BRPOP）是一个阻止列表流行原语。您可以将此命令视为阻止 LPOP 版本，如果指定的键不存在或包含空列表，则能够阻止 RPOP。
            // 我们描述了 BLPOP，但这两个命令是相同的，唯一的区别是 BLPOP 从列表的左侧（头部）弹出元素，而 BRPOP 从右侧（尾部）弹出。
            // 调用 BLPOP 时，如果至少有一个指定的键包含非空列表，则会从列表的头部弹出一个元素
            // 并与键的名称一起返回给调用方（BLPOP 返回一个双元素数组，第一个元素是键，第二个元素是弹出的值）
            for (String item : jedis.brpop(0,"javacui")) {
                // 读取到相关数据，进行业务处理
                System.out.println("接收到消息：" + item);
            }
        }
    }
}

END

推荐您阅读更多有关于“ list 队列 redis 消息 堵塞 ”的文章

猜你喜欢

• 3-8HTTP POST数据和数据接收
• 3-8常用SpringJPA主键生成策略，UUID和自增
• 3-8Apache Commons Jar包下载
• 3-8IIS服务器返回Bad Request (Invalid Hostname)解决办法
• 3-8SpringJPA 慎重 findAll() 切记分页