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

文章目录

Spring Boot+Redis 实现一个轻量级的消息队列
0.前言
1.基础介绍
2.步骤
- 2.1. 引入依赖
- 2.2. 配置文件
- 2.3. 核心源码
4.总结
- 答疑
5.参考文档
6. 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.1
    port: 6379
    password: your_password
    database: 0
  redis.jedis.pool:
    max-active: 50
    max-idle: 10
    min-idle: 5
    max-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 {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String receivedMessage = message.toString();
        // 处理接收到的消息
        System.out.println("Received message: " + receivedMessage);
    }
}

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

@Configuration
public class RedisConfig {

    @Bean
    public ChannelTopic topic() {
        return new ChannelTopic("messageQueue"); // 定义消息队列的通道名称
    }

    @Bean
    public MessageListenerAdapter messageListenerAdapter(MessageSubscriber messageSubscriber) {
        return new MessageListenerAdapter(messageSubscriber);
    }

    @Bean
    public 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.参考文档

Redis官方文档 - 发布/订阅：
- 链接：https://redis.io/topics/pubsub ↗
- 该文档提供了Redis发布/订阅功能的详细说明，包括基本概念、命令使用、示例代码和常见问题解答等。
Redis命令文档 - PUBLISH：
- 链接：https://redis.io/commands/publish ↗
- 该文档介绍了PUBLISH命令的语法、参数和用法，以及与发布消息相关的注意事项。
Redis命令文档 - SUBSCRIBE：
- 链接：https://redis.io/commands/subscribe ↗
- 该文档详细说明了SUBSCRIBE命令的语法、参数和用法，以及如何订阅频道并接收发布的消息。
Redis命令文档 - UNSUBSCRIBE：
- 链接：https://redis.io/commands/unsubscribe ↗
- 该文档提供了UNSUBSCRIBE命令的说明，包括取消订阅频道的语法和参数。
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 实现消息队列实践示例，全部内容就是这些。如果你有疑问或见解可以在评论区留言。