使用Redis发布订阅模式实现 Session共享

其实并不是实现session共享，而是通过redis的发布订阅，让所有集群的服务器，都让自己的session发送一下消息。比如说userId在第35台服务器上， 有100台服务器，那么第1台服务器收到消息，需要通知userId，不是找到第35台服务器，而是通知所有的服务器，给userId发条消息，其他99台服务器没有userId，那就发送不成功！

1、配置redis

package com.kakarote.crm.config;import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.kakarote.crm.constant.RedisConstants;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.connection.lettuce.LettucePoolingClientConfiguration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;import java.time.Duration;@Configuration
public class CrmTemplateConfig {@Value("${spring.redis.host}")private String redisHost;@Value("${spring.redis.port}")private int redisPort;@Value("${spring.redis.password}")private String redisHasrdpwd;@Value("${spring.redis.database}")private Integer database;@Bean(name = "crmRedisTemplate")public RedisTemplate redisTemplate() {RedisTemplate<String, Object> template = new RedisTemplate<>();template.setConnectionFactory(connectionFactory(database, redisHost, redisPort, redisHasrdpwd));template.setKeySerializer(new StringRedisSerializer());template.setValueSerializer(new GenericJackson2JsonRedisSerializer());template.setHashKeySerializer(new StringRedisSerializer());template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());return template;}public RedisConnectionFactory connectionFactory(int database, String hostName, int port, String password) {RedisStandaloneConfiguration configuration = new RedisStandaloneConfiguration();configuration.setHostName(hostName);configuration.setPort(port);if (StringUtils.isNotBlank(password)) {configuration.setPassword(password);}if (database != 0) {configuration.setDatabase(database);}GenericObjectPoolConfig genericObjectPoolConfig = new GenericObjectPoolConfig();genericObjectPoolConfig.setMaxIdle(10);genericObjectPoolConfig.setMinIdle(10);genericObjectPoolConfig.setMaxTotal(100);genericObjectPoolConfig.setMaxWaitMillis(3000);LettuceClientConfiguration clientConfig = LettucePoolingClientConfiguration.builder().commandTimeout(Duration.ofMillis(8000)).poolConfig(genericObjectPoolConfig).build();LettuceConnectionFactory lettuce = new LettuceConnectionFactory(configuration, clientConfig);lettuce.afterPropertiesSet();return lettuce;}/*** Redis消息监听器容器* 这个容器加载了RedisConnectionFactory和消息监听器* 可以添加多个监听不同话题的redis监听器，只需要把消息监听器和相应的消息订阅处理器绑定，该消息监听器* 通过反射技术调用消息订阅处理器的相关方法进行一些业务处理** @return redis消息监听容器*/@Bean@SuppressWarnings("all")public RedisMessageListenerContainer container(RedisMessageListener listener) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();// 监听所有库的key过期事件container.setConnectionFactory(connectionFactory(database, redisHost, redisPort, redisHasrdpwd));// 所有的订阅消息，都需要在这里进行注册绑定,new PatternTopic(TOPIC_NAME1)表示发布的主题信息// 可以添加多个 messageListener，配置不同的通道container.addMessageListener(listener, new PatternTopic(RedisConstants.WEBSOCKET_REDIS_TOPIC));/*** 设置序列化对象* 特别注意：1. 发布的时候需要设置序列化；订阅方也需要设置序列化*         2. 设置序列化对象必须放在[加入消息监听器]这一步后面，否则会导致接收器接收不到消息*/Jackson2JsonRedisSerializer seria = new Jackson2JsonRedisSerializer(Object.class);ObjectMapper objectMapper = new ObjectMapper();objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);seria.setObjectMapper(objectMapper);container.setTopicSerializer(seria);return container;}
}

2、配置RedisMessageListener

package com.kakarote.crm.config;import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import com.kakarote.crm.constant.CrmConst;
import com.kakarote.crm.entity.BO.MessageDto;
import com.kakarote.crm.websocket.TransferCallWebSocket;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;@Slf4j
@Component
public class RedisMessageListener implements MessageListener {@Autowiredprivate RedisTemplate<String, Object> crmRedisTemplate;@Overridepublic void onMessage(Message message, byte[] pattern) {// 接收的topiclog.info("RedisMessageListener-接收到消息1，channel:" + new String(pattern));try {//序列化对象（特别注意：发布的时候需要设置序列化；订阅方也需要设置序列化）MessageDto messageDto = (MessageDto) crmRedisTemplate.getValueSerializer().deserialize(message.getBody());log.info("RedisMessageListener-接收到消息2，channel = {}, messageDto = {}", new String(pattern), messageDto);if(messageDto == null){log.info("RedisMessageListener-messageDto = null，无消息进行发送! message = {}", JSONUtil.toJsonStr(message));return;}if(CrmConst.NOTICE_MSG.equals(messageDto.getTitle())){JSONObject content = messageDto.getContent();String toUserId = content.getString("toUserId");String fromUserId = content.getString("fromUserId");JSONObject msg = content.getJSONObject("msg");String resp = TransferCallWebSocket.sendMsgByUserId(fromUserId, toUserId, JSONUtil.toJsonStr(msg));if(!resp.equals("success")){log.info("RedisMessageListener-发送弹框消息,resp = {},content = {}", resp, content);}}}catch (Exception e){log.info("RedisMessageListener-监听消息处理失败，失败原因 = {}, e = ", e.getMessage(), e);}}
}

3、静态类

/*** @description: 常量类* @dateTime: 2021/6/17 16:21*/
public class RedisConstants {/*** UTF-8 字符集*/public static final String UTF8 = "UTF-8";public final static String WEBSOCKET_REDIS_TOPIC = "websocket_topic";public static final String TRANSFER_NOTICE = "transferCallNotice";	public static final String NOTICE_MSG = "noticeMessage";
}

4、消息体

package com.kakarote.crm.entity.BO;import com.alibaba.fastjson.JSONObject;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;import java.io.Serializable;@AllArgsConstructor
@NoArgsConstructor
@Data
public class MessageDto implements Serializable {private String data;private String title;private JSONObject content;
}

5、业务类像通道发送消息

    /*** 向通道发布消息*/public boolean convertAndSend(String channel, Object message) {if (StringUtil.isBlank(channel)) {return false;}try {crmRedisTemplate.convertAndSend(channel, message);log.info("发送消息成功，channel：{}，message：{}", channel, message);return true;} catch (Exception e) {log.info("发送消息失败，channel：{}，message：{}, 失败原因 = {}, e = ", channel, message, e.getMessage(), e);e.printStackTrace();}return false;}

6、websocket配置

@Configuration
@ComponentScan
@EnableAutoConfiguration
public class WebSocketConfiguration implements ServletContextInitializer {@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}@Beanpublic TaskScheduler taskScheduler(){ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();taskScheduler.setPoolSize(10);taskScheduler.initialize();return taskScheduler;}@Overridepublic void onStartup(ServletContext servletContext) throws ServletException {servletContext.addListener(WebAppRootListener.class);servletContext.setInitParameter("org.apache.tomcat.websocket.textBufferSize","52428800");servletContext.setInitParameter("org.apache.tomcat.websocket.binaryBufferSize","52428800");}
}

7、websocket Controller类

@ServerEndpoint("/crmDzhWebsocket/transferWebsocket/{userId}")
@Component
@Slf4j
public class TransferCallWebSocket {/*** 当前在线连接数*/private static AtomicInteger onlineCount = new AtomicInteger(0);/*** 用来存放每个客户端对应的 WebSocketServer 对象*/private static final ConcurrentHashMap<String, Session> webSocketMap = new ConcurrentHashMap<>();/*** 与某个客户端的连接会话，需要通过它来给客户端发送数据*/private Session session;/*** 接收 userId*/private String userIdKey = "";/*** 连接建立成功调用的方法*/@OnOpenpublic void onOpen(Session session, @PathParam("userId") String userId) {this.session = session;this.userIdKey = userId;if (webSocketMap.containsKey(userId)) {webSocketMap.remove(userId);webSocketMap.put(userId, session);} else {webSocketMap.put(userId, session);addOnlineCount();}log.info("转接通知用户连接:" + userId + ",当前总在线人数为:" + getOnlineCount());try {sendMessage("success");} catch (IOException e) {log.error("转接通知用户:" + userId + ",网络异常!!!!!!");log.info("转接通知用户连接:" + userId + ",网络异常!!!!!!");}}/*** 连接关闭调用的方法*/@OnClosepublic void onClose() {if (webSocketMap.containsKey(userIdKey)) {webSocketMap.remove(userIdKey);subOnlineCount();}log.info("转接通知用户退出:" + userIdKey + ",当前总在线人数为:" + getOnlineCount());}/*** 收到客户端消息后调用的方法** @param message 客户端发送过来的消息*/@OnMessagepublic void onMessage(String message, Session session) {try {if ("ping".equals(message)) {webSocketMap.get(this.userIdKey).getBasicRemote().sendText("pong");return;}log.info("this.userIdKey = {}, message = {}", this.userIdKey, message);} catch (IOException e) {log.error("转接通知发送消息失败，失败原因 = {}, e = ", e.getMessage(), e);e.printStackTrace();}}public static String sendMsgByUserId(String fromUserId, String toUserId, String msg) throws IOException {if(webSocketMap.get(toUserId) != null){try {webSocketMap.get(toUserId).getBasicRemote().sendText(msg);return "success";}catch (Exception e){log.error("发送消息失败,fromUserId = {}, toUserId = {}", fromUserId, toUserId);return e.getMessage();}}return "userId:" + toUserId + "当前不在会话中";}/*** 发生错误时调用** @param session* @param error*/@OnErrorpublic void onError(Session session, Throwable error) {log.info("用户错误:" + session.getId() + ",原因:" + error.getMessage());}/*** 实现服务器主动推送*/public void sendMessage(String message) throws IOException {this.session.getBasicRemote().sendText(message);}public static synchronized AtomicInteger getOnlineCount() {return onlineCount;}public static synchronized void addOnlineCount() {TransferCallWebSocket.onlineCount.getAndIncrement();}public static synchronized void subOnlineCount() {TransferCallWebSocket.onlineCount.getAndDecrement();}}