Redis系列学习文章分享---第七篇（Redis快速入门之消息队列--List实现消息队列 Pubsub实现消息队列 stream的单消费模式 stream的消费者组模式 基于stream消息队列）

Jedis jedis = new Jedis("localhost", 6379);
// 生产者：将消息推入队列
jedis.lpush("message_queue", "message1");
jedis.lpush("message_queue", "message2");
// 消费者：从队列尾部获取消息
String message = jedis.rpop("message_queue");
System.out.println("Received message: " + message);

Jedis jedis = new Jedis("localhost", 6379);
// 发布消息
jedis.publish("channel", "message");
// 订阅消息
JedisPubSub subscriber = new JedisPubSub() {
    @Override
    public void onMessage(String channel, String message) {
        System.out.println("Received message: " + message);
    }
};
jedis.subscribe(subscriber, "channel");

Jedis jedis = new Jedis("localhost", 6379);
// 添加消息到Stream
Map<String, String> message = new HashMap<>();
message.put("id", "1");
message.put("content", "message1");
jedis.xadd("message_stream", StreamEntryID.NEW_ENTRY, message);
// 读取消息
List<EntryID> messageIds = jedis.xread(StreamEntryID.LAST_ENTRY, StreamEntryID.NEW_ENTRY,
                                        StreamEntryID.UNRECEIVED_ENTRY, 1);
for (EntryID messageId : messageIds) {
    Map<String, String> message = jedis.hgetAll(messageId.toString());
    System.out.println("Received message: " + message);
}

Jedis jedis = new Jedis("localhost", 6379);
// 创建消费者组
jedis.xgroupCreate("message_stream", "consumer_group", StreamEntryID.NEW_ENTRY, true);
// 消费消息
List<EntryID> messageIds = jedis.xreadGroup("consumer_group", "consumer", StreamEntryID.LAST_ENTRY, 1, false);
for (EntryID messageId : messageIds) {
    Map<String, String> message = jedis.hgetAll(messageId.toString());
    System.out.println("Received message: " + message);
    // 处理消息
    jedis.xack("message_stream", "consumer_group", messageId);
}

Jedis jedis = new Jedis("localhost", 6379);
// 添加秒杀请求到Stream
Map<String, String> seckillRequest = new HashMap<>();
seckillRequest.put("id", "1");
seckillRequest.put("productId", "1001");
seckillRequest.put("userId", "10001");
jedis.xadd("seckill_requests", StreamEntryID.NEW_ENTRY, seckillRequest);
// 消费秒杀请求
List<EntryID> requestIds = jedis.xreadGroup("seckill_group", "consumer", StreamEntryID.LAST_ENTRY, 1, false);
for (EntryID requestId : requestIds) {
    Map<String, String> request = jedis.hgetAll(requestId.toString());
    System.out.println("Received seckill request: " + request);
    // 处理秒杀请求，如扣减库存等操作
    // 处理完成后，确认消息已处理
    jedis.xack("seckill_requests", "seckill_group", requestId);
}