Websocket通过Redis实现Session共享

pom依赖

	 
 
     org.springframework.boot
     spring-boot-starter-data-redis
 

 
 
     org.springframework.boot
     spring-boot-starter-websocket
 

开启Websocket配置

@Configuration
@EnableWebSocket
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
 return new ServerEndpointExporter();
    }
}

WebsocketPool类

package com.chainter.rmblc.messaging.net;

import lombok.extern.java.Log;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.websocket.Session;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;


@Log
public class WebSocketPool {
    // 当前在线人数
    private static final AtomicInteger atomicNumber = new AtomicInteger(0);
    // 当前Websocket session连接
    private static Map onlineSession = new ConcurrentHashMap<>();


    public static Integer addAtomicNumber(){
 return atomicNumber.incrementAndGet();
    }
    public static Integer decrementNumber(){
 return atomicNumber.decrementAndGet();
    }
    public static Integer getNumber(){
 return atomicNumber.get();
    }

    public static void createonlineSession(String userId,Session session){
 onlineSession.put(userId,session);
    }
    public static Map getonlineSession(){
 return onlineSession;
    }
    public static Session getSesssionByUserId(String userId){
 return Optional.ofNullable(onlineSession.get(userId)).orElse(null);
    }
    public static void removeSession(String userId){
 Session session = onlineSession.get(userId);
 if(ObjectUtils.isEmpty(session)){
     return;
 }
 try {
     session.close();
     onlineSession.remove(userId);
 } catch (IOException e) {
     log.warning("关闭连接出现错误");
 }
    }

    public static void send(){
 onlineSession.values().stream().forEach(session -> {
     try {
  Date date = new Date();
  long time = date.getTime();
  String dateString = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date);
  session.getBasicRemote().sendText("时间："+dateString+",毫秒："+time);
     } catch (Exception e) {
  e.printStackTrace();
     }
 });
    }

}

ws连接类

@Component
@ServerEndpoint(value = "/WebSocketTest/{userId}")
@Log
public class WebsocketTest {


    @onOpen
    public void onOpen(@PathParam("userId") String userId, Session session) {

 Integer number = WebSocketPool.addAtomicNumber();
 WebSocketPool.createonlineSession(session.getId(),session);
// MessageTaskHandle.createTaskHandle(session,"服务器主动推送信息");
 log.info("建立连接，当前人数："+number);
    }

    @onClose
    public void onClose(@PathParam("userId") String userId, Session session){
 WebSocketPool.removeSession(session.getId());
 Integer number = WebSocketPool.decrementNumber();
 log.info("用户"+userId+"关闭连接,当前人数："+number);
    }

    @onError
    public void onError(@PathParam("userId") String userId, Session session,Throwable throwable){
 WebSocketPool.removeSession(session.getId());
 log.warning("WebSocket连接出现异常");
    }



}

添加redis监听类

package com.chainter.rmblc.messaging.config;

import com.chainter.rmblc.messaging.handle.RedisListenerHandle;
import lombok.extern.java.Log;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.stereotype.Component;


@Component
@Log
public class RedisListenerBean {
		 
		 // application.yml中配置allWSName
    @Value("${sub.channel.allWSName}")
    private String allWSName;

    
    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter) {
 RedisMessageListenerContainer container = new RedisMessageListenerContainer();
 container.setConnectionFactory(connectionFactory);

 // 监听msgToAll
 container.addMessageListener(listenerAdapter, new PatternTopic(allWSName));
 log.info("Subscribed Redis channel: " + allWSName);
 return container;
    }

    @Bean
    public MessageListenerAdapter messageListenerAdapter(RedisListenerHandle redisListenerHandle){
 return new MessageListenerAdapter(redisListenerHandle,"receiveMessage");
    }
}

创建RedisListenerHandle监听消息处理类

package com.chainter.rmblc.messaging.handle;

import org.springframework.stereotype.Component;


@Component
public class RedisListenerHandle {

    public void receiveMessage(String message){
 System.out.println("接收消息："+message);
    }

}