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

Redis的实战篇-消息队列

本文将介绍Redis中消息队列的使用实践，包括基于List和Pub/Sub实现的消息队列，以及最新引入的Stream数据结构的应用场景。通过学习本文，您将了解如何使用Redis构建高效的消息队列系统，并掌握在实际项目中应用消息队列的技巧。

目录

1. 认识消息队列
2. 基于List实现消息队列
3. Pub/Sub实现消息队列
4. Stream的单消费模式
5. Stream的消费者组模式
6. 基于Stream消息队列实现异步秒杀

1. Redis消息队列-认识消息队列

1.1 什么是消息队列

消息队列是一种在应用程序之间传递消息的通信方式。它通常用于解耦消息的发送者和接收者，提高系统的可靠性、扩展性和性能。

1.2 消息队列的作用

• 异步处理：将耗时的任务放入消息队列中，由后台异步处理，提高系统的响应速度。
• 解耦应用：通过消息队列，不同模块之间可以解耦，降低模块之间的耦合度。
• 削峰填谷：在高峰期，消息队列可以缓冲大量请求，避免系统崩溃或响应变慢。

2. Redis消息队列-基于List实现消息队列

2.1 使用List结构

Redis的List结构可以很方便地实现消息队列，使用LPUSH命令将消息推入队列头部，使用RPOP命令从队列尾部获取消息。

2.2 示例代码

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);
1
2
3
4
5
6
7
8
9

3. Redis消息队列-Pub/Sub实现消息队列

3.1 使用Pub/Sub模式

Redis的Pub/Sub模式通过订阅者（Subscriber）和发布者（Publisher）实现消息的传递。订阅者订阅某个频道，发布者向该频道发布消息，订阅者接收到消息并处理。

3.2 示例代码

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");
1
2
3
4
5
6
7
8
9
10
11
12
13

4. Redis消息队列-Stream的单消费模式

4.1 使用Stream数据结构

Redis 5.0引入了Stream数据结构，可以用于实现消息队列。在单消费模式下，每个消息只能被一个消费者处理。

4.2 示例代码

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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

5. Redis消息队列-Stream的消费者组模式

5.1 消费者组概念

在消费者组模式下，多个消费者可以共同消费同一个Stream，每个消息只会被消费一次，从而实现负载均衡和高可用性。

5.2 示例代码

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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13

6. Redis消息队列-基于stream消息队列实现异步秒杀

6.1 异步秒杀的概念

异步秒杀是一种常见的电商场景，在高并发情况下，为了保证系统的稳定性和性能，通常采用异步方式处理秒杀请求，将请求放入消息队列中，由消费者逐个处理。

6.2 示例代码

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);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

以上是基于Redis Stream消息队列实现异步秒杀的示例代码，通过消息队列的方式，可以有效减轻系统压力，提高系统的并发处理能力。

结语

本文介绍了Redis消息队列的基本概念和常见实现方式，并通过示例代码演示了如何使用Redis Stream实现消息队列。希望本文能够帮助您更好地理解和应用Redis消息队列技术。如果您对本文有任何疑问或建议，请随时在评论区留言，感谢您的阅读！