3 redis实现一个消息中间件

使用list实现一个队列，可以从左侧入队，也可以从右侧入对
即可以从左侧读取，也可以从右侧读取

1、Lindex

Lindex 命令用于通过索引获取列表中的元素
也可以使用负数下标，以 -1 表示列表的最后一个元素， -2 表示列表的倒数第二。

redis 127.0.0.1:6379> LINDEX KEY_NAME INDEX_POSITION 
1

列表中下标为指定索引值的元素。 如果指定索引值不在列表的区间范围内，返回 nil。
例如：

redis 127.0.0.1:6379> LPUSH mylist "World"
(integer) 1
 
redis 127.0.0.1:6379> LPUSH mylist "Hello"
(integer) 2
 
redis 127.0.0.1:6379> LINDEX mylist 0
"Hello"
 
redis 127.0.0.1:6379> LINDEX mylist -1
"World"
 
redis 127.0.0.1:6379> LINDEX mylist 3        # index不在 mylist 的区间范围内
(nil)
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2、Rpush

Rpush 命令用于将一个或多个值插入到列表的尾部(最右边)
如果列表不存在，一个空列表会被创建并执行 RPUSH 操作。 当列表存在但不是列表类型时，返回一个错误。

redis 127.0.0.1:6379> RPUSH KEY_NAME VALUE1..VALUEN
1

返回值：是执行 RPUSH 操作后，列表的长度。
例如：

redis 127.0.0.1:6379> RPUSH mylist "hello"
(integer) 1
redis 127.0.0.1:6379> RPUSH mylist "foo"
(integer) 2
redis 127.0.0.1:6379> RPUSH mylist "bar"
(integer) 3
redis 127.0.0.1:6379> LRANGE mylist 0 -1
1) "hello"
2) "foo"
3) "bar"
1
2
3
4
5
6
7
8
9
10

set是做不了的，set没有顺序，无法满足FIFO

3、代码实现

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.List;

@Component
public class ListVer{
    public final static String RS_LIST_MQ_NS = "rlm:";

    @Autowired
    private JedisPool jedisPool;

    /*消费者接受消息*/
    public List<String> get(String key) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            //阻塞式的等待消息，如果队列中没有数据的话，就一直等
            //0：表示等待0s,没有数据就不等待了
            return jedis.brpop(0,RS_LIST_MQ_NS + key);
        } catch (Exception e) {
            throw new RuntimeException("接受消息失败！");
        } finally {
            jedis.close();
        }
    }

    /*生产者发送消息*/
    public void put(String key, String message) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.lpush(RS_LIST_MQ_NS+key,message);
        } catch (Exception e) {
            throw new RuntimeException("发送消息失败！");
        } finally {
            jedis.close();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

注意：保持key的一致性

延迟消息的场景：
比如已经加入购物车中，购买的电影票，还没有付款的时候锁定座位，进行延迟消息投递。
之后再启动另一个消费端去监控这个延迟消息，到了时间后还没有支付的话，则进行取消操作.

可以使用sorted set（有序集合）来实现延迟消息，实际生产中不建议使用，还是使用正经的MQ。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.TimeUnit;

@Component
public class ZSetVer {
    //标记是延迟消息的key
    public final static String RS_ZS_MQ_NS = "rzsm:";

    @Autowired
    private JedisPool jedisPool;

    /*生产者，消息的发送，实际生产中，相关参数，
    比如订单信息，过期时间等应该传入，可以考虑将订单信息json化存入redis*/
    public void producer() {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            for (int i = 0; i < 5; i++) {
                //订单id
                String order_id = "000000000"+i;
                double score = System.currentTimeMillis()+(i*1000);
                jedis.zadd(RS_ZS_MQ_NS+"orderId",score, order_id);
                System.out.println("生产订单: " + order_id + " 当前时间："
                        + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
                System.out.println((3 + i) + "秒后执行");
            }
        } catch (Exception e) {
            throw new RuntimeException("生产消息失败！");
        } finally {
            jedis.close();
        }

    }

    //消费者，取订单
    public void consumerDelayMessage() {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            while (true) {
                Set<String> order = jedis.zrangeByScore(RS_ZS_MQ_NS+"orderId", 0,
                        System.currentTimeMillis(), 0,1);
                if (order == null || order.isEmpty()) {
                    System.out.println("当前没有等待的任务");
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    continue;
                }
                String s = order.iterator().next();

                if (jedis.zrem(RS_ZS_MQ_NS+"orderId", s)>0) {
                    /*业务处理*/
                    System.out.println(s);
                }
            }

        } catch (Exception e) {
            throw new RuntimeException("消费消息失败！");
        } finally {
            jedis.close();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73