springboot整合websocket

2021-01-28 10:01:04 阅读：171 来源： 互联网

标签：websocket springboot userId session 整合 message public String

一、引入maven依赖

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

二、websocket配置类

@Configuration
public class WebSocketConfig {


    /**
     * ServerEndpointExporter 作用
     *
     * 这个Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint
     *
     * @return
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

三、websocket业务代码

 1 @Slf4j
 2 @Component
 3 @ServerEndpoint("/websocket/{userId}")
 4 public class WebSocket {
 5 
 6     /**
 7      *  与某个客户端的连接对话，需要通过它来给客户端发送消息
 8      */
 9     private Session session;
10 
11     /**
12      * 标识当前连接客户端的用户id
13      */
14     private String userId;
15 
16     /**
17      *  用于存所有的连接服务的客户端，这个对象存储是安全的
18      */
19     private static ConcurrentHashMap<String,WebSocket> webSocketSet = new ConcurrentHashMap<>();
20 
21 
22     /**
23      * 开启连接
24      * @param session
25      * @param userId
26      */
27     @OnOpen
28     public void onOpen(Session session, @PathParam(value = "userId") String userId){
29         this.session = session;
30         this.userId = userId;
31         // userId是用来表示唯一客户端，如果需要指定发送，需要指定发送通过userId来区分
32         webSocketSet.put(userId,this);
33         log.info("[WebSocket] 连接成功，当前连接人数为：={}",webSocketSet.size());
34     }
35 
36     /**
37      * 退出连接
38      */
39     @OnClose
40     public void onClose(){
41         if (webSocketSet.containsKey(this.userId)){
42             webSocketSet.remove(this.userId);
43             log.info("[WebSocket] 退出成功，当前连接人数为：={}",webSocketSet.size());
44         }
45     }
46 
47     /**
48      * 收到客户端消息后调用的方法
49      * @param message
50      */
51     @OnMessage
52     public void onMessage(String message, Session session) throws IOException {
53        log.info("收到消息：{}-{}",session,message);
54        session.getBasicRemote().sendText("收到了消息："+message);
55     }
56 
57 
58     /**
59      * 发送自定义消息
60      * */
61     public static void sendInfo(String message,@PathParam("userId") String userId) {
62         if(StringUtils.isNotBlank(userId) && webSocketSet.containsKey(userId)){
63             try {
64                 webSocketSet.get(userId).session.getBasicRemote().sendText(message);
65             } catch (IOException e) {
66                log.error("消息发送失败：{}-{}-{}",userId,message,e.getMessage());
67             }
68         }else{
69             log.error("用户{},不在线！",userId);
70         }
71     }

View Code

四、使用redis实现websocket的session共享

由于websocket的session不能序列化，所以不能直接存储到redis中（而且就算可以存储到redi中，A系统产生的session可能也不能在B系统使用），因此可以采用rredis发布/订阅配置方式实现共享。

4.1、redis发布/订阅配置

@Configuration
@EnableCaching
public class RedisConfig {

    public static final  String TOPIC_PATTERN = "channel:websocket";

    @Autowired
    private RedisConnectionFactory connectionFactory;

    @Bean
    RedisMessageListenerContainer container(MessageListenerAdapter listenerAdapter) {

        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        // 可以添加多个 messageListener，配置不同的交换机
        container.addMessageListener(listenerAdapter, new PatternTopic(TOPIC_PATTERN));
        return container;
    }

    @Bean
    MessageListenerAdapter listenerAdapter(RedisReceiver receiver) {
       //监听消息的对象和方法
        return new MessageListenerAdapter(receiver, "onMessage");
    }
}

4.2、监听类

@Slf4j
@Component
public class RedisReceiver implements MessageListener {
 // 字符串的分隔符
    public static final String DELIMITER ="✈";

    @Override
    public void onMessage(Message message, byte[] bytes) {
        byte[] body = message.getBody();
        if (body.length>0){
            String content = new String(body);
            if (content.contains(DELIMITER)){
                String[] split = StringUtils.split(content, DELIMITER);
                if (split.length == 2){
                    WebSocket.sendInfo(split[1],split[0]);
                }
            }
        }
    }
}

4.3、编写测试接口

    @ApiOperation("测试webSocket")
    @PostMapping("testWebSocket")
    public R testWebSocket(@ApiParam("发送的消息") @RequestParam String message, @ApiParam("接收用户id")@RequestParam String userId){
        redisTemplate.convertAndSend(RedisConfig.TOPIC_PATTERN,userId+ RedisReceiver.DELIMITER+message);
        return R.ok();
    }

备注：websocket的在线测试地址：http://www.websocket-test.com/

标签：websocket,springboot,userId,session,整合,message,public,String
来源： https://www.cnblogs.com/cq-yangzhou/p/14338211.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9