Spring Boot+Redis 实现一个轻量级的消息队列

0.前言

本文将介绍如何利用Spring Boot与Redis结合实现消息队列的实践示例。消息队列是一种常用的解耦和异步通信的机制，可以在系统中实现高效的消息传递和处理。而Redis作为一种高性能的内存数据库，具备发布/订阅功能，非常适合用于构建消息队列系统。

在本实践示例中，我们将使用Spring Boot框架和Redis作为基础，通过Redis的发布/订阅模型来实现简单的消息队列。 如何通过消息通道进行消息的发布和订阅。

在开始之前，你需要对Spring Boot和Redis有一定的了解，并确保已正确配置好开发环境。本文将提供实际的代码示例和详细的步骤说明， 并能够根据具体业务需求进行扩展和定制。

1.基础介绍

Redis提供了发布/订阅（Publish/Subscribe）模式，用于实现消息的广播和异步通信。下面是关于Redis发布/订阅模式的介绍：

发布/订阅模式（Pub/Sub）：
发布/订阅模式是一种消息通信模式，其中消息的发布者（Publisher）将消息发送到特定的频道（Channel），而订阅者（Subscriber）可以订阅一个或多个频道以接收消息。这种模式允许消息的广播和异步传递，发送者和接收者之间解耦。

Redis的发布/订阅功能：
Redis提供了原生的发布/订阅功能，使得开发者可以使用Redis作为消息中间件来实现高效的消息传递。以下是与Redis发布/订阅相关的关键概念和操作：
下面是将Redis发布/订阅功能整理成表格的示例：

概念	说明
频道（Channel）	Redis中用于发布和订阅消息的通道。每个消息都被发布到一个特定的频道，而订阅者可以选择订阅一个或多个感兴趣的频道。
发布消息（Publish）	通过使用PUBLISH命令，发布者可以将消息发送到指定的频道。一旦消息被发布到频道，所有订阅该频道的客户端将接收到该消息。
订阅频道（Subscribe）	通过使用SUBSCRIBE命令，客户端可以订阅一个或多个频道。一旦订阅成功，客户端将成为该频道的订阅者，可以接收到该频道上发布的消息。
取消订阅频道（Unsubscribe）	通过使用UNSUBSCRIBE命令，客户端可以取消对一个或多个频道的订阅。当客户端不再对某个频道感兴趣时，可以选择取消订阅。
模式匹配订阅（Pattern Subscription）	Redis支持使用PSUBSCRIBE命令进行模式匹配订阅。通过指定一个模式，可以订阅与该模式匹配的多个频道。

通过Redis的发布/订阅功能，可以构建高效的消息系统，实现消息的广播和异步传递。发布者将消息发布到特定的频道，而订阅者可以选择订阅感兴趣的频道来接收消息。这种模式的优点包括解耦、异步通信和实时性等，适用于很多场景，如实时通知、事件驱动系统和分布式系统间的消息传递等。

本文我们通过Spring Boot +Redis 实现一个轻量级的消息队列。

2.步骤

2.1. 引入依赖

<dependencies><!-- Spring Data Redis --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency>
</dependencies>

2.2. 配置文件

# Redis连接配置
spring.redis.host=127.0.0.1
spring.redis.port=6379
spring.redis.password=your_password
spring.redis.database=0# Redis连接池配置
spring.redis.jedis.pool.max-active=50
spring.redis.jedis.pool.max-idle=10
spring.redis.jedis.pool.min-idle=5
spring.redis.jedis.pool.max-wait=-1

在上面的配置中，您可以根据实际情况修改以下属性：

• spring.redis.host：Redis服务器的主机名或IP地址。
• spring.redis.port：Redis服务器的端口号。
• spring.redis.password：Redis服务器的密码（如果有的话）。
• spring.redis.database：Redis数据库的索引，默认为0。

另外，您还可以配置Redis连接池的属性，以控制连接池的行为。在示例配置中，设置了以下连接池属性：

• spring.redis.jedis.pool.max-active：连接池中的最大活动连接数。
• spring.redis.jedis.pool.max-idle：连接池中的最大空闲连接数。
• spring.redis.jedis.pool.min-idle：连接池中的最小空闲连接数。
• spring.redis.jedis.pool.max-wait：从连接池获取连接的最大等待时间（毫秒），-1表示无限等待。

如果 使用的是YAML格式的配置文件（application.yml），可以将上述配置转换为相应的格式：

spring:redis:host: 127.0.0.1port: 6379password: your_passworddatabase: 0redis.jedis.pool:max-active: 50max-idle: 10min-idle: 5max-wait: -1

请根据您的实际Redis服务器配置进行调整，并根据需要添加其他相关配置，如超时设置、SSL配置等。

2.3. 核心源码

当使用Spring Boot与Redis结合实现消息队列时，可以使用Redis的发布/订阅功能来处理消息的发布和订阅。下面是

接下来，创建一个消息发布者和一个消息订阅者的类。在发布者类中，使用RedisTemplate来发布消息，而在订阅者类中，通过实现MessageListener接口来处理接收到的消息。

// 消息发布者
@Component
public class MessagePublisher {private final RedisTemplate<String, String> redisTemplate;private final ChannelTopic topic;public MessagePublisher(RedisTemplate<String, String> redisTemplate, ChannelTopic topic) {this.redisTemplate = redisTemplate;this.topic = topic;}public void publishMessage(String message) {redisTemplate.convertAndSend(topic.getTopic(), message);}
}

消息订阅者

// 消息订阅者
@Component
public class MessageSubscriber implements MessageListener {@Overridepublic void onMessage(Message message, byte[] pattern) {String receivedMessage = message.toString();// 处理接收到的消息System.out.println("Received message: " + receivedMessage);}
}

接下来，配置Redis消息监听容器，以便启动消息监听器：

@Configuration
public class RedisConfig {@Beanpublic ChannelTopic topic() {return new ChannelTopic("messageQueue"); // 定义消息队列的通道名称}@Beanpublic MessageListenerAdapter messageListenerAdapter(MessageSubscriber messageSubscriber) {return new MessageListenerAdapter(messageSubscriber);}@Beanpublic RedisMessageListenerContainer redisContainer(RedisConnectionFactory redisConnectionFactory,MessageListenerAdapter messageListenerAdapter,ChannelTopic topic) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(redisConnectionFactory);container.addMessageListener(messageListenerAdapter, topic);return container;}
}

最后，在需要发布消息的地方，注入MessagePublisher并调用publishMessage()方法来发布消息：

@Service
public class MyService {private final MessagePublisher messagePublisher;public MyService(MessagePublisher messagePublisher) {this.messagePublisher = messagePublisher;}public void doSomethingAndPublishMessage() {// 执行一些操作String message = "Hello, Redis message queue!";messagePublisher.publishMessage(message);}
}

这样，当调用MyService中的doSomethingAndPublishMessage()方法时，消息将被发布到Redis的消息队列中，并由订阅者接收和处理。

请注意，以上示例是一个简化的实现，仅用于演示基本的消息发布和订阅流程。在实际的生产环境中，可能需要考虑更多的细节，如错误处理、消息确认机制、并发处理等。

4.总结

通过Redis的发布/订阅模型，Spring Boot利用RedisTemplate将消息发布到指定的Redis通道，同时使用MessageListenerAdapter将订阅者与消息通道绑定。当有消息发布到通道时，Redis会将消息广播给所有订阅该通道的客户端，然后MessageListenerAdapter将接收到的消息传递给相应的消息监听器进行处理。

这样，利用Redis的发布/订阅功能，我们可以实现基于Spring Boot与Redis的消息队列系统，实现消息的发布和订阅。通过这种异步的方式，消息发布者和订阅者可以解耦，提高系统的可伸缩性和性能。

答疑

上次介绍了 Redis 的pub和sub，以及Stream 有很多同学有疑惑项目中应该使用哪种。那么我们在此处再聊聊
Redis提供了两种不同的功能来实现消息队列：发布/订阅（Pub/Sub）和流（Stream）。选择使用哪种功能作为消息队列取决于您的具体需求和使用场景。

特点	Redis发布/订阅（Pub/Sub）	Redis流（Stream）
适用场景	简单的消息广播和异步通信场景	更复杂的消息队列需求，包括任务队列、事件溯源、日志收集等
消息持久化	不支持消息的持久化和历史记录	支持消息的持久化和历史记录
消息订阅方式	订阅者选择订阅感兴趣的频道来接收消息	多个消费者对消息进行消费
订阅前发布的消息	无法获取先于订阅的消息	可以获取订阅前发布的消息
实时性	适用于需要即时通知、实时事件处理和实时数据传递的场景	实时性较高，但受消费者的处理速度影响
消费者组管理	不支持消费者组管理	支持消费者组管理
消息顺序性	消息的顺序不保证	消息的顺序保证
用例示例	即时聊天应用、实时通知系统、发布/订阅模式的消息传递	任务队列、事件溯源、日志收集等

所以如果你的需求是简单的消息广播和异步通信，并且对消息的持久化和历史记录要求不高，可以选择使用Redis发布/订阅功能作为消息队列。而如果您需要更丰富的消息队列功能，包括消息的持久化、多个消费者的支持和消息历史记录等，那么Redis流是更好的选择。根据具体的使用场景和需求，选择适合的功能来实现消息队列。

5.参考文档

1. Redis官方文档 - 发布/订阅：

   • 链接：https://redis.io/topics/pubsub ↗
   • 该文档提供了Redis发布/订阅功能的详细说明，包括基本概念、命令使用、示例代码和常见问题解答等。

2. Redis命令文档 - PUBLISH：

   • 链接：https://redis.io/commands/publish ↗
   • 该文档介绍了PUBLISH命令的语法、参数和用法，以及与发布消息相关的注意事项。

3. Redis命令文档 - SUBSCRIBE：

   • 链接：https://redis.io/commands/subscribe ↗
   • 该文档详细说明了SUBSCRIBE命令的语法、参数和用法，以及如何订阅频道并接收发布的消息。

4. Redis命令文档 - UNSUBSCRIBE：

   • 链接：https://redis.io/commands/unsubscribe ↗
   • 该文档提供了UNSUBSCRIBE命令的说明，包括取消订阅频道的语法和参数。

5. Redis命令文档 - PSUBSCRIBE：

   • 链接：https://redis.io/commands/psubscribe ↗
   • 该文档介绍了PSUBSCRIBE命令的用法，用于进行模式匹配订阅，可以订阅与指定模式匹配的多个频道。

6. Redis从入门到精通系列文章

• 《Redis使用Lua脚本和Redisson来保证库存扣减中的原子性和一致性》
• 《SpringBoot Redis 使用Lettuce和Jedis配置哨兵模式》
• 《Redis【应用篇】之RedisTemplate基本操作》
• 《Redis 从入门到精通【实践篇】之SpringBoot配置Redis多数据源》
• 《Redis 从入门到精通【进阶篇】之三分钟了解Redis HyperLogLog 数据结构》
• 《Redis 从入门到精通【进阶篇】之三分钟了解Redis地理位置数据结构GeoHash》
• 《Redis 从入门到精通【进阶篇】之高可用哨兵机制(Redis Sentinel)详解》
• 《Redis 从入门到精通【进阶篇】之redis主从复制详解》
• 《Redis 从入门到精通【进阶篇】之Redis事务详解》
• 《Redis从入门到精通【进阶篇】之对象机制详解》
• 《Redis从入门到精通【进阶篇】之消息传递发布订阅模式详解》
• 《Redis从入门到精通【进阶篇】之持久化 AOF详解》
• 《Redis从入门到精通【进阶篇】之持久化RDB详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构字典(Dictionary)详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构快表QuickList详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构简单动态字符串(SDS)详解》
• 《Redis从入门到精通【高阶篇】之底层数据结构压缩列表(ZipList)详解》
• 《Redis从入门到精通【进阶篇】之数据类型Stream详解和使用示例》
  大家好，我是冰点，今天的 Spring Boot+Redis 实现消息队列实践示例，全部内容就是这些。如果你有疑问或见解可以在评论区留言。