redis（五）---- 简单消息队列

消息队列一个消息的链表，是一个异步处理的数据处理引擎。不仅能够提高系统的负荷，还能够改善因网络阻塞导致的数据缺失。一般用于邮件发送、手机短信发送，数据表单提交、图片生成、视频转换、日志储存等。

redis的list类型天生支持用作消息队列。由于redis的list是使用双向链表实现的，保存了头尾节点，所以在列表头尾两边插取元素都是非常快的。所以可以直接使用redis的list实现消息队列，只需简单的两个指令lpush和rpop或者rpush和lpop。简单示例如下：

public class MessageQueue {

    private final String redisChanel1 = "redisChanel1";
    private final String redisChanel2 = "redisChanel2";

    private String redisHost = "10.5.31.155";
    private int redisPort = 6379;
    private Jedis redis;

    @Before
    public void before() {
        redis = new Jedis(redisHost, redisPort);
    }

    @Test
    public void pubChanel1() throws InterruptedException {
        for (int i = 0; i < 1000; i++) {
            redis.lpush(redisChanel1, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 第" + i + "条消息");
            Thread.sleep(Math.round(Math.floor(Math.random() * 2000)));
        }
    }

    @Test
    public void pubChanel2() throws InterruptedException {
        for (int i = 0; i < 1000; i++) {
            redis.lpush(redisChanel2, new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 第" + i + "条消息");
            Thread.sleep(Math.round(Math.floor(Math.random() * 2000)));
        }
    }

    @Test
    public void sub() throws InterruptedException {
        while (true) {
            String message1 = redis.rpop(redisChanel1);
            System.out.println(redisChanel1 + "-->" + message1);
            Thread.sleep(200);
            String message2 = redis.rpop(redisChanel2);
            System.out.println(redisChanel2 + "-->" + message2);
            Thread.sleep(200);
        }
    }

    @After
    public void after() {
        redis.close();
    }
}

先后运行pubChanel1、pubChanel2、sub方法，可以看到类似如下输出：

redisChanel1-->2018-09-28 10:23:44 第0条消息
redisChanel2-->2018-09-28 10:23:46 第0条消息
redisChanel1-->2018-09-28 10:23:44 第1条消息
redisChanel2-->2018-09-28 10:23:47 第1条消息
redisChanel1-->2018-09-28 10:23:46 第2条消息
redisChanel2-->null
redisChanel1-->2018-09-28 10:23:47 第3条消息
redisChanel2-->2018-09-28 10:23:49 第2条消息
redisChanel1-->2018-09-28 10:23:48 第4条消息
redisChanel2-->null
redisChanel1-->2018-09-28 10:23:49 第5条消息

以上实现原理非常简单，但是由于消费者取消息是循环+sleep实现的，所以会出现如下问题：

需要考虑消费者拿到空消息的情况（输出结果中出现了null）。
如果生产者速度大于消费者消费速度，消息队列长度会一直增大，时间久了会占用大量内存空间。
消费者存在资源浪费的情况。

索性redis还支持另外一个命令：brpop，这个命令只有当list中有元素的时候才会返回，没有元素时会阻塞，直到阻塞超时返回null。那么改变下sub方法的写法如下：

@Test
public void sub() {
    while (true) {
        List<String> messages = redis.brpop(0, redisChanel1, redisChanel2);
        for (String message : messages) {
            System.out.println(message);
        }
    }
}

其中，brpop中的第一个参数值为0表示无超时时间，没有取到消息的时候一直阻塞下去。

redis（五）---- 简单消息队列

相关推荐