整合redis消息隊列

項目依賴

<!-- RedisTemplate --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency><!-- Redis-Jedis --><dependency><groupId>redis.clients</groupId><artifactId>jedis</artifactId><version>2.9.0</version></dependency>

application.yml配置文件配置redis信息

spring:# Redis配置redis:timeout: 10slettuce:pool:max-active: 200max-idle: 8max-wait: 10smin-idle: 2shutdown-timeout: 3sdatabase: 0port: 6379host: 127.0.0.1password:

配置redis監(jiān)聽器（配置訂閱的頻道）

package com.sinosoft.springbootplus.common.config;import com.sinosoft.springbootplus.common.service.MySubscribe; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.redis.connection.RedisConnectionFactory; import org.springframework.data.redis.listener.PatternTopic; import org.springframework.data.redis.listener.RedisMessageListenerContainer;/*** redis配置* @author lsh* @date 2022/8/9*/ @Configuration public class RedisConfig {/*** redis消息監(jiān)聽器容器*/@Beanpublic RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);//訂閱頻道，通配符*表示任意多個占位符container.addMessageListener(new MySubscribe(), new PatternTopic("websocket"));return container;} }

接收訂閱的消息

1、此處是接收訂閱的消息，然后處理訂閱的消息，處理與redis消息本身沒有關系。所以可以轉換成對象，再調具體處理的接口（處理的接口不需要關心消息本身）
2、存入redis消息隊列時，用了genericJackson2JsonRedisSerializer的序列化方式，需要使用genericJackson2JsonRedisSerializer的反序列化方式反序列化成對象。

package com.sinosoft.springbootplus.common.service;import com.sinosoft.springbootplus.common.api.DealMySubscribeMessageApi; import com.sinosoft.springbootplus.common.vo.WebSocketMessageVo; import com.sinosoft.springbootplus.util.MapperUtils; import com.sinosoft.springbootplus.util.SpringContextUtil; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.data.redis.connection.Message; import org.springframework.data.redis.connection.MessageListener; import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;/*** 我的訂閱* @author lsh* @date 2022/8/9*/ @Slf4j public class MySubscribe implements MessageListener {private DealMySubscribeMessageApi dealMySubscribeMessageApi;GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();@SneakyThrows@Overridepublic void onMessage(Message message, byte[] bytes) {if(null == dealMySubscribeMessageApi){dealMySubscribeMessageApi = SpringContextUtil.getBean(DealMySubscribeMessageApi.class);}log.debug("訂閱頻道是【{}】接收的數(shù)據(jù)是【{}】",new String(message.getChannel()),new String(message.getBody()));String channel = new String(message.getChannel());String macAddress = StringUtils.substring(channel,9);//此處將訂閱的消息統(tǒng)一轉換成了對象，后面處理具體的消息與redis消息本身沒有關系WebSocketMessageVo webSocketMessageVo =(WebSocketMessageVo) (genericJackson2JsonRedisSerializer.deserialize(message.getBody()));dealMySubscribeMessageApi.dealMySubscribeMessage(macAddress,webSocketMessageVo);} }

處理訂閱的消息api

package com.sinosoft.springbootplus.common.api;import com.sinosoft.springbootplus.common.vo.WebSocketMessageVo; import org.springframework.data.redis.connection.Message;/*** 處理我訂閱的消息接口* @author lsh* @date 2022/8/9*/ public interface DealMySubscribeMessageApi {/*** 處理我訂閱的消息*/void dealMySubscribeMessage(String macAddress, WebSocketMessageVo webSocketMessageVo); }

實現(xiàn)處理訂閱的消息api接口

package com.sinosoft.springbootplus.common.service;import com.alibaba.fastjson.JSONObject; import com.sinosoft.springbootplus.common.api.DealMySubscribeMessageApi; import com.sinosoft.springbootplus.common.vo.RedisMessageVo; import com.sinosoft.springbootplus.common.vo.WebSocketMessageVo; import com.sinosoft.springbootplus.system.vo.SessionVo; import com.sinosoft.springbootplus.util.Jackson; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.data.redis.connection.Message; import org.springframework.stereotype.Service;/*** 處理我訂閱的消息實現(xiàn)類* @author lsh* @date 2022/8/9*/ @Service @Slf4j public class DealMySubscribeMessageImpl implements DealMySubscribeMessageApi {private MyWebsocketImpl myWebsocket;public DealMySubscribeMessageImpl(MyWebsocketImpl myWebsocket) {this.myWebsocket = myWebsocket;}@Overridepublic void dealMySubscribeMessage(String macAddress,WebSocketMessageVo webSocketMessageVo) {log.info("給設備【{}】發(fā)消息【{}】",macAddress,Jackson.toJsonString(webSocketMessageVo));myWebsocket.send(Jackson.toJsonString(webSocketMessageVo),macAddress);} }

總結

以上是生活随笔為你收集整理的springboot：整合redis消息队列的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。