SpringBoot+Redis 实现消息订阅发布

31次阅读

共计 2213 个字符，预计需要花费 6 分钟才能阅读完成。

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

Redis 发布订阅

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

SpringBoot+Redis 实现消息订阅发布

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

功能实现

发布消息

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

SpringBoot+Redis 实现消息订阅发布

订阅频道

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

SpringBoot+Redis 实现消息订阅发布

模式匹配

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

SpringBoot+Redis 实现消息订阅发布

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

取消订阅

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

SpringBoot+Redis 实现发布订阅

pringboot 如何整合 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());
    }
}

运行结果

SpringBoot+Redis 实现消息订阅发布