Java实现基于Redis的发布/订阅
一、场景使用Redis实现消息的发布订阅,由生产者(pub)发布消息,消费者(sub)消费消息。这里有个 channel 的概念,就是一个通道,消费者订阅这个通道,然后发布者在这个通道发布消息,订阅这个通道的消费者都可以消费到。二、命令发布消息# channel是指通道,message是指消息的内容,按照这个格式,可以给该通道发布消息PUBLISH channel message# 比如要往一个叫
·
一、场景
使用Redis实现消息的发布订阅,由生产者(pub)发布消息,消费者(sub)消费消息。
这里有个 channel 的概念,就是一个通道,消费者订阅这个通道,然后发布者在这个通道发布消息,订阅这个通道的消费者都可以消费到。
二、命令
发布消息
# channel是指通道,message是指消息的内容,按照这个格式,可以给该通道发布消息
PUBLISH channel message
# 比如要往一个叫 demo的通道里发个消息 helloworld
PUBLISH demo helloworld
消费消息
# channel 是指要订阅的通道,这里要和消息发布者的通道保持一致即可收到消息
SUBSCRIBE channel
#比如我要订阅上面创建的通道
SUBSCRIBE demo
三、Java实现
使用Java来实现的话,有下面步骤就可以简单实现了,一个是配置RedisMessageListenerContainer 消息监听容器,然后配置消息监听适配器MessageListenerAdapter,最后就是创建一个监听的类,然后写个方法来实现监听回调。
需要导入的依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
1、配置消息监听
/**
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/29 9:06
*/
@Configuration
public class RedisConfig {
//这里我写在配置文件里了,可以自定义,只要和发布消息时定义的channel保持一致就可以
@Value("${redis.topic.test}")
private String topic;
@Bean
public RedisMessageListenerContainer listenerContainer(RedisConnectionFactory factory,
MessageListenerAdapter messageListenerAdapter){
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
container.addMessageListener(messageListenerAdapter, new PatternTopic(topic));
return container;
}
@Bean
public MessageListenerAdapter messageListenerAdapter(RedisReceiver redisReceiver){
MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(redisReceiver,
"onMessage");
return messageListenerAdapter;
}
}
2、编写消息监听时回调的方法
/**
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/29 9:10
*/
@Component
@Slf4j
public class RedisReceiver implements MessageListener {
@Override
public void onMessage(Message message, byte[] bytes) {
System.out.println("进来了回调方法!");
log.info("回调的信息 message={}", message);
log.info("回调的信息 bytes={}", bytes);
}
}
3、发送消息
/**
* @author linzp
* @version 1.0.0
* CreateDate 2021/1/29 9:20
*/
@Service
public class PublishService {
//这里我使用stringRedisTemplate
@Autowired
private StringRedisTemplate stringRedisTemplate;
//这里使用convertAndSend方法,参数为 channel 通道和内容
public void sendMsg(String channel, String msg){
stringRedisTemplate.convertAndSend(channel, msg);
}
}
四、测试
启动项目,然后开启本地的一个客户端,发布消息
127.0.0.1:6379> PUBLISH SEND_MSG helloworld
(integer) 1
127.0.0.1:6379> ^C
这时查看项目的控制台,发现输出日志如下,即调用成功
码字不易,如果文章帮助了你,点个赞吧~
更多推荐
已为社区贡献6条内容
所有评论(0)