基于java实现redis实现发布与订阅 |
您所在的位置：网站首页 › 订阅与发布 › 基于java实现redis实现发布与订阅 |
基于java实现redis实现发布与订阅
|
目录 三、Redis的发布和订阅 三、发布和订阅的实现 四、发布和订阅命令 1、subscribe：订阅一个或者多个频道 2、publish：发布消息到指定的频道 3、psubscribe：订阅一个或多个符合给定模式的频道 4、pubsub：查看订阅与发布系统状态 5、punsubscribe：退订所有给定模式的频道 6、unsubscribe：指退订给定的频道 五、Redis发布和订阅缺点 六、redis当中实现发布和订阅
一、发布和订阅介绍 官网介绍：Redis 发布/订阅是一种消息传模式，其中发送者（在Redis术语中称为发布者）发送消息，而接收者（订阅者）接收消息。传递消息的通道称为channel。 redis的发布与订阅缺点：发布订阅的消息不会被持久化，所以新订阅客户端不会接收到历史消息。 二：常用命令： 命令描述PSUBSCRIBE订阅一个或多个符合给定模式的频道。PUBSUB查看订阅与发布系统状态。PUBLISH将信息发送到指定的频道。PUNSUBSCRIBE退订所有给定模式的频道。SUBSCRIBE订阅给定的一个或多个频道的信息。UNSUBSCRIBE指退订给定的频道。 三、Redis的发布和订阅下图展示了client1，client2，client5订阅了channel1频道。
下图展示了消息发送给客户端： 当有新消息通过 publish命令发送给channel1 时， 这个消息就会发送给订阅了该频道的客户端。
1：第一个redis客户端： 订阅命令： subscribe 频道1 频道2 … ，可以订阅多个频道。当执行命令过后只要不停止会一直处于订阅监听状态。 127.0.0.1:6379> SUBSCRIBE channel01 Reading messages... (press Ctrl-C to quit) 1) "subscribe" 2) "channel01" 3) (integer) 12、打开另一个客户端，给channel1发布消息hello 发标消息命令： publish channel 消息 ，返回值表示有几个订阅者 redis 127.0.0.1:6379> PUBLISH channel01 "hello" (integer) 1订阅者客户端就会收到信息： 127.0.0.1:6379> SUBSCRIBE channel01 Reading messages... (press Ctrl-C to quit) 1) "message" 2) "channel01" 3) "hello" 四、发布和订阅命令 1、subscribe：订阅一个或者多个频道 2、publish：发布消息到指定的频道PUBLISH channel 发送的消息 3、psubscribe：订阅一个或多个符合给定模式的频道PSUBSCRIBE pattern [pattern ...] 4、pubsub：查看订阅与发布系统状态pubsub channels [pattern] 5、punsubscribe：退订所有给定模式的频道对应的是psubscribe。 PUNSUBSCRIBE 如果没有参数，那么客户端使用 PSUBSCRIBE 命令订阅的所有模式都会被退订。 PUNSUBSCRIBE 如果没有参数，那么只会退订给定参数的客户端 6、unsubscribe：指退订给定的频道unsubscribe: 如果没有指定频道，那么所有频道都会被退订 。 五、Redis发布和订阅缺点1：发布订阅的消息不会被持久化 2: 当客户端发送消息：如果一个消费者都没有，那么消息直接丢弃。如果开始有一个或者多个消费者，一个消费者突然down掉了，生产者会继续发送消息，其他消费者可以持续收到消息。但是挂掉的消费者重新连上的时候，生产者发送的消息，对于down掉的消费者来说就是彻底丢失了。 六、redis当中实现发布和订阅1：创建top配置类 @Configuration public class TopicConfig { //订阅发布的主题 @Bean ChannelTopic topic() { return new ChannelTopic( "topic:queue" ); } }2: 注册消息监听容器 @Configuration public class SubListenerConfig { //消息监听 @Bean MessageListenerAdapter messageListener() { return new MessageListenerAdapter( new MessageListener() ); } //消息监听容器 /* RedisMessageListenerContainer 其实现了InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle几个接口 InitializingBean：主要实现afterPropertiesSet方法，来定义spring设置完properties后进行的处理，在spring init这个bean时候会被调用 DisposableBean：实现destroy方法，在spring销毁bean时会调用 BeanNameAware：实现setBeanName方法来为bean进行取名，在RedisMessageListenerContainer中该name被用于内部线程的线程名 SmartLifecycle：spring的bean生命周期类，spring会调用start，stop等操作来完成RedisMessageListenerContainer类的启动 */ @Bean RedisMessageListenerContainer redisContainer(RedisConnectionFactory factory) { final RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(factory); container.addMessageListener(messageListener(), new ChannelTopic( "topic:queue" )); return container; } }3：配置监听器 @Component public class MessageListener implements MessageListener { @Override public void onMessage(Message message, byte[] bytes) { System.out.println("Message received: " + message.toString()); } }4：发送消息端配置 @Component public class MessageListener { private Logger logger = LoggerFactory.getLogger(MessageListener.class); @Autowired private RedisTemplate redisTemplate; @Autowired private ChannelTopic topic; //发送消息 public void sendMsg(String msg){ redisTemplate.convertAndSend( topic.getTopic(), "Message==:" + msg + ";Time:" + Calendar.getInstance().getTime()); } }5：创建controller，发送消息 @Controller public class RedisTestController { @Autowired private MessageListener messageListener; @RequestMapping("/send") @ResponseBody public String sendMsg(@RequestParam("message") String message){ messageListener.sendMsg(msg); return "消息发送成功！"; } } |
【本文地址】
今日新闻 |
推荐新闻 |