当前位置：首页 > news >正文

3 redis实现一个消息中间件

news 来源：原创 2024/5/14 11:19:36

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

1、Lindex

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

redis 127.0.0.1:6379> LINDEX KEY_NAME INDEX_POSITION

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

redis 127.0.0.1:6379> LPUSH mylist "World"
(integer) 1redis 127.0.0.1:6379> LPUSH mylist "Hello"
(integer) 2redis 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)

2、Rpush

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

redis 127.0.0.1:6379> RPUSH KEY_NAME VALUE1..VALUEN

返回值：是执行 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"

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:";@Autowiredprivate 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();}}
}

注意：保持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 {//标记是延迟消息的keypublic final static String RS_ZS_MQ_NS = "rzsm:";@Autowiredprivate JedisPool jedisPool;/*生产者，消息的发送，实际生产中，相关参数，比如订单信息，过期时间等应该传入，可以考虑将订单信息json化存入redis*/public void producer() {Jedis jedis = null;try {jedis = jedisPool.getResource();for (int i = 0; i < 5; i++) {//订单idString 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();}}
}