SpringBoot+Redis 实现消息订阅发布

2022-11-03 08:34:32

什么是 Redis

Redis 是一个开源的使用 ANSI C语言编写的内存数据库，它以 key-value 键值对的形式存储数据，高性能，读取速度快，也提供了持久化存储机制。

Redis 通常在项目中用的最多的功能是缓存，然而今天靓仔将为你介绍它的另一个功能，轻量级的消息队列。

Redis 发布订阅

机制

Redis 提供了发布订阅功能，可以用于消息的传输，Redis 的发布订阅机制包括三个部分，发布者，订阅者和 Channel（称之为频道或主题）。

发布者将消息发布到指定频道，订阅该频道的订阅者就能够接受到这条消息，如果有多个订阅者，那么他们会接受到相同的消息。

功能实现

发布消息

Redis 采用 PUBLISH 命令发送消息，其返回值为接收到该消息的订阅者的数量。

订阅频道

Redis 采用 SUBSCRIBE 命令订阅某个频道，其返回值包括客户端订阅的频道，目前已订阅的频道数量，以及接收到的消息，其中subscribe表示已经成功订阅了某个频道。

模式匹配

模式匹配功能允许客户端订阅符合某个模式的频道，Redis采用PSUBSCRIBE订阅符合某个模式所有频道，用“ * ”表示模式，“ * ”可以被任意值代替。

假设客户端同时订阅了某种模式和符合该模式的某个频道，那么发送给这个频道的消息将被客户端接收到两次，只不过这两条消息的类型不同，一个是 message 类型，一个是 pmessage 类型，但其内容相同。

取消订阅

Redi s采用 UNSUBSCRIBE 和 PUNSUBSCRIBE 命令取消订阅，其返回值与订阅类似。由于Redis的订阅操作是阻塞式的，因此一旦客户端订阅了某个频道或模式，就将会一直处于订阅状态直到退出。在 SUBSCRIBE，PSUBSCRIBE，UNSUBSCRIBE 和 PUNSUBSCRIBE 命令中，其返回值都包含了该客户端当前订阅的频道和模式的数量，当这个数量变为0时，该客户端会自动退出订阅状态。

SpringBoot+Redis 实现发布订阅

springboot 如何整合 redis 我这里就不讲了，相信对你来说也没有一点问题。我们直接上代码

消息监听配置

@Configuration

public class RedisSubConfig {

    public static final String SUB_KEY = "chat";//频道channel

    /**

     * redis消息监听器容器

     * 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器

     * 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理

     * @param connectionFactory

     * @param listenerAdapter

     * @return

     */

    @Bean

    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();

        container.setConnectionFactory(connectionFactory);

        //订阅了一个频道

        container.addMessageListener(listenerAdapter, new PatternTopic(RedisSubConfig.SUB_KEY));

        return container;

    }

    /**

     * 消息监听器适配器，绑定消息处理器，利用反射技术调用消息处理器的业务方法

     * @param receiver

     * @return

     */

    @Bean

    MessageListenerAdapter listenerAdapter(RedisReceiver receiver) {

        return new MessageListenerAdapter(receiver, "receiveMessage");

    }

    /**

     * redis 读取内容的template

     * @param connectionFactory

     * @return

     */

    @Bean

    StringRedisTemplate template(RedisConnectionFactory connectionFactory) {

        return new StringRedisTemplate(connectionFactory);

    }

}

接收消息

@Service

public class RedisReceiver {

    public void receiveMessage(String message) {

        System.out.println("接收消息：" + message);

    }

}

采用定时器发布消息

@EnableScheduling //开启定时器功能

@Component

public class MessageSender {

    @Autowired

    private StringRedisTemplate stringRedisTemplate;

    @Scheduled(fixedRate = 5000) //间隔5s 通过StringRedisTemplate对象向redis消息队列chat频道发布消息

    public void sendMessage(){

        stringRedisTemplate.convertAndSend("chat", "hello "+ new Date());

    }

}