SpringBoot Redis 發布訂閱模式 Pub/Sub
注意：redis的發布訂閱模式不可以將消息進行持久化，訂閱者發生網絡斷開、宕機等可能導致錯過消息。
Redis命令行下使用發布訂閱
publish 發布
發布者通過以下命令可以往指定channel發布message
redis> publish channel message
subscribe 訂閱
訂閱者通過以下命令可以訂閱一個或多個頻道，如果頻道不存在則會創建
redis> subscribe channel [channel ...]
對於redis的發布訂閱的命令就這么簡單。那么接下來我們在springboot中如何使用發布訂閱的功能呢？
SpringBoot中使用Redis的發布訂閱功能
添加依賴配置redis信息和連接池什么的就不說了，如果添加的有commons-pool2依賴的話，會自動幫我們配置redis連接池的
發布者
相對於訂閱者來說，發布者的實現方式很簡單，以下方式就可以往channel中發送message了。
@Resource
private RedisTemplate<String, Object> redisTemplate;
public void publish(){
// 使用高級的redisTemplate
redisTemplate.convertAndSend("channel","message");
// 使用低級的connection 實際上redisTemplate的底層就是使用的下面的方式
redisTemplate.execute(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
connection.publish("channel".getBytes(StandardCharsets.UTF_8), "message".getBytes(StandardCharsets.UTF_8));
return null;
}
}, true);
// true這個參數意思是 是否將redis連接暴露給回調代碼，大多數情況下設置true就可以了，往后深入的話可以看到
RedisConnection connToExpose = (exposeConnection ? connToUse : createRedisConnectionProxy(connToUse)); 如果為false的話會創建redis連接的代理
}
訂閱者
訂閱者因為涉及到連接、線程等 所以內容相對會多一點
@Resource
private RedisTemplate<String, Object> redisTemplate;
public void subscribe() {
redisTemplate.execute(new RedisCallback<Object>() {
@Override
public Object doInRedis(RedisConnection connection) throws DataAccessException {
// 我定義了一個全局的 ConcurrentHashMap 用來存放連接 因為后面的取消訂閱的線程要和訂閱的線程用同一個連接
map.put("connection",connection);
// subscribe 按頻道訂閱 該方法會阻塞該線程 只有取消訂閱才會釋放該線程
connection.subscribe(new MessageListener() {
@Override
public void onMessage(Message message, byte[] pattern) {
log.info("接收到消息");
System.out.println(new String(message.getBody()));
}
}, "channelOne".getBytes(StandardCharsets.UTF_8), "channelTwo".getBytes(StandardCharsets.UTF_8));
// 按模式訂閱 pSubscribe 只有取消訂閱才會釋放該線程
// connection.pSubscribe(new MessageListener() {
// @Override
// public void onMessage(Message message, byte[] pattern) {
// System.out.println(new String(message.getBody()));
// }
// }, "patternOne".getBytes(StandardCharsets.UTF_8), "patternOne".getBytes(StandardCharsets.UTF_8));
return null;
}
}, true);
}
如何取消訂閱呢？從剛才的map里取到連接
RedisConnection the = map.get("connection");
Subscription subscription = the.getSubscription();
subscription.unsubscribe();
消息監聽容器
上面的那種訂閱為低級訂閱，由於連接在調用subscribe的時候會導致當前線程阻塞，這種方式需要對每個監聽器連接和線程管理，所以spring提供了RedisMessageListenerContainer類來幫我們完成這些工作。
RedisMessageListenerContainer顧名思義可以知道它是一個消息監聽容器
詳情請參考官方文檔
如何實現
@Configuration
public class DefaultMessageListenerContainerConfig {
@Bean
public RedisMessageListenerContainer container(RedisConnectionFactory factory) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
// 官方推薦我們使用自定義的線程池或者使用TaskExecutor
container.setTaskExecutor(executor());
container.addMessageListener(new MessageListener() {
@Override
public void onMessage(Message message, byte[] pattern) {
System.out.println(Thread.currentThread().getName() + ": " + new String(message.getBody()));
}
}, new ChannelTopic("message"));
return container;
}
@Bean
public TaskExecutor executor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);
executor.setQueueCapacity(100);
executor.initialize();
return executor;
}
}
這個時候我們在redis命令行內使用 publish channel message 的時候，我們的spring程序就可以訂閱到消息了。
再說下 MessageListenerAdapter
我們可以通過 MessageListenerAdapter 消息接收者包裝進去，消息接收者不會和redis有任何耦合。
官方文檔給了spring傳統的xml的方式配置的，下面我給出基於configuration配置的代碼
public interface MessageDelegate {
void handleMessage(String message);
}
public class DefaultMessageDelegate implements MessageDelegate {
@Override
public void handleMessage(String message) {
System.out.println(message);
}
}
@Configuration
public class MessageListenerContainerConfig {
@Autowired
private DefaultMessageDelegate defaultMessageDelegate;
@Bean
public RedisMessageListenerContainer container(RedisConnectionFactory factory,
MessageListenerAdapter messageListenerAdapter) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
container.setTaskExecutor(executor());
Map<MessageListenerAdapter, Collection<? extends Topic>> map = new HashMap<>();
List<ChannelTopic> channelTopics = new ArrayList<>();
ChannelTopic channelTopic = new ChannelTopic("message");
channelTopics.add(channelTopic);
map.put(messageListenerAdapter, channelTopics);
container.setMessageListeners(map);
return container;
}
@Bean
public TaskExecutor executor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(Runtime.getRuntime().availableProcessors());
executor.setMaxPoolSize(Runtime.getRuntime().availableProcessors() * 2);
executor.setQueueCapacity(100);
executor.initialize();
return executor;
}
@Bean
public MessageListenerAdapter messageListenerAdapter() {
// handleMessage 參數消息來的時候要調用的方法 默認是 handleMessage
return new MessageListenerAdapter(defaultMessageDelegate, "handleMessage");
}
}
如果我們要在程序運行時添加訂閱或者取消訂閱的時候該怎么辦呢？
我們需要提前准備好消息偵聽器，添加的時候把偵聽器注入到消息容器
取消的時候就調用消息容器的remove方法把偵聽器刪除掉即可。