当前位置：首页 > news >正文

即时通讯代码优化

news 2026/4/3 6:10:44

在线用户逻辑修复

在进行测试时，发现当前代码有个问题，如果test1在服务器进行连接，本地的test2给test1发消息，虽然test1能收到服务器上的信息，但是本地服务日志中会报teset1不在线，需要对该种情况进行修复

修复方案：使用redis存储在线用户

WebSocketEndpoint

利用setBit记录登录用户,key为用户名的hashcode，即便有可能冲突，但是概率较小，可以接受。

package com.example.im.endpoint;import com.example.im.app.service.WebSocketMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;/*** @author PC*/
@Component
@ServerEndpoint("/ws")
public class WebSocketEndpoint {private final static Logger logger = LoggerFactory.getLogger(WebSocketEndpoint.class);public static final ConcurrentHashMap<Integer, WebSocketEndpoint> WEB_SOCKET_ENDPOINT_MAP = new ConcurrentHashMap<>();/*** redis中用户SessionMap*/public static final String ONLINE_USER = "cus:ws:online-user";private Session session;private static WebSocketMessageService webSocketMessageService;private static RedisTemplate<String, String> redisTemplate;@Autowiredpublic void setWebSocketMessageService(WebSocketMessageService webSocketMessageService) {WebSocketEndpoint.webSocketMessageService = webSocketMessageService;}@Autowiredpublic void setRedisTemplate(RedisTemplate<String, String> redisTemplate) {WebSocketEndpoint.redisTemplate = redisTemplate;}/*** 打开ws连接** @param session 会话*/@OnOpenpublic void onOpen(Session session) {//连接成功String userName = getUserName(session);logger.info("The connection is successful:" + getUserName(session));this.session = session;//是有hash冲突的可能性的，不过触发概率很低，可以忽略int hashCode = userName.hashCode();redisTemplate.opsForValue().setBit(ONLINE_USER, hashCode, true);WEB_SOCKET_ENDPOINT_MAP.put(hashCode, this);}/*** 断开ws连接** @param session 会话*/@OnClosepublic void onClose(Session session) {String userName = getUserName(session);int hashCode = userName.hashCode();redisTemplate.opsForValue().setBit(ONLINE_USER, hashCode, false);WEB_SOCKET_ENDPOINT_MAP.remove(hashCode);//断开连接logger.info("Disconnect:" + userName);}/*** 接收到的消息** @param message 消息内容*/@OnMessagepublic void onMessage(String message, Session session) {//接收消息String sendUserName = getUserName(session);webSocketMessageService.sendMessage(sendUserName, message);}private String getUserName(Session session) {return Optional.ofNullable(session.getRequestParameterMap().get("userName")).orElse(new ArrayList<>()).stream().findFirst().orElse("anonymous_users");}public Session getSession() {return session;}public void setSession(Session session) {this.session = session;}
}

DefaultSendExecutor

适配改变类型后的WEB_SOCKET_ENDPOINT_MAP，并调整代码结构

package com.example.im.endpoint;import com.example.im.app.service.WebSocketMessageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;/*** @author PC*/
@Component
@ServerEndpoint("/ws")
public class WebSocketEndpoint {private final static Logger logger = LoggerFactory.getLogger(WebSocketEndpoint.class);public static final ConcurrentHashMap<Integer, WebSocketEndpoint> WEB_SOCKET_ENDPOINT_MAP = new ConcurrentHashMap<>();/*** redis中用户SessionMap*/public static final String ONLINE_USER = "cus:ws:online-user";private Session session;private static WebSocketMessageService webSocketMessageService;private static RedisTemplate<String, String> redisTemplate;@Autowiredpublic void setWebSocketMessageService(WebSocketMessageService webSocketMessageService) {WebSocketEndpoint.webSocketMessageService = webSocketMessageService;}@Autowiredpublic void setRedisTemplate(RedisTemplate<String, String> redisTemplate) {WebSocketEndpoint.redisTemplate = redisTemplate;}/*** 打开ws连接** @param session 会话*/@OnOpenpublic void onOpen(Session session) {//连接成功String userName = getUserName(session);logger.info("The connection is successful:" + getUserName(session));this.session = session;//是有hash冲突的可能性的，不过触发概率很低，可以忽略int hashCode = userName.hashCode();redisTemplate.opsForValue().setBit(ONLINE_USER, hashCode, true);WEB_SOCKET_ENDPOINT_MAP.put(hashCode, this);}/*** 断开ws连接** @param session 会话*/@OnClosepublic void onClose(Session session) {String userName = getUserName(session);int hashCode = userName.hashCode();redisTemplate.opsForValue().setBit(ONLINE_USER, hashCode, false);WEB_SOCKET_ENDPOINT_MAP.remove(hashCode);//断开连接logger.info("Disconnect:" + userName);}/*** 接收到的消息** @param message 消息内容*/@OnMessagepublic void onMessage(String message, Session session) {//接收消息String sendUserName = getUserName(session);webSocketMessageService.sendMessage(sendUserName, message);}private String getUserName(Session session) {return Optional.ofNullable(session.getRequestParameterMap().get("userName")).orElse(new ArrayList<>()).stream().findFirst().orElse("anonymous_users");}public Session getSession() {return session;}public void setSession(Session session) {this.session = session;}
}

重复代码提取

在增加了stream和redis渠道后，消息监听处的代码有大段重复，可以进行提取处理

MessageInfoUtil

package com.example.im.infra.executor.send.util;import com.example.im.infra.executor.send.DefaultSendExecutor;
import com.example.im.infra.executor.send.dto.MessageInfo;
import com.example.im.infra.util.JsonUtils;
import com.fasterxml.jackson.core.type.TypeReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import java.nio.charset.StandardCharsets;/*** @author PC* MessageInfo工具类*/
@Component
public class MessageInfoUtil {private final static Logger logger = LoggerFactory.getLogger(MessageInfoUtil.class);private DefaultSendExecutor defaultSendExecutor;@Autowiredpublic void setDefaultSendExecutor(DefaultSendExecutor defaultSendExecutor) {this.defaultSendExecutor = defaultSendExecutor;}public MessageInfo sendMessageByMessageInfoByte(byte[] messageInfoByte) {String messageJson = new String(messageInfoByte, StandardCharsets.UTF_8);MessageInfo messageInfo = JsonUtils.toObjectByTypeReference(messageJson, new TypeReference<MessageInfo>() {});switch (messageInfo.getScopeOfSending()) {case USER:defaultSendExecutor.sendToUser(messageInfo.getSendUserName(), messageInfo.getMessage());break;case ALL:defaultSendExecutor.sendToAll(messageInfo.getSendUserName(), messageInfo.getMessage());break;default://一般来说不会出现该情况，除非用户覆盖了ScopeOfSending，后续可以开个扩展发送范围的口子logger.warn("invalid sending range:" + messageInfo.getScopeOfSending().getScopeCode());break;}return messageInfo;}
}

RedisMessageListener

package com.example.im.infra.executor.send.redis;import com.example.im.infra.executor.send.util.MessageInfoUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;/*** @author PC* redis监听*/
@Component
public class RedisMessageListener implements MessageListener {private final static Logger logger = LoggerFactory.getLogger(RedisMessageListener.class);private MessageInfoUtil messageInfoUtil;@Autowiredpublic void setMessageInfoUtil(MessageInfoUtil messageInfoUtil) {this.messageInfoUtil = messageInfoUtil;}@Overridepublic void onMessage(Message message, byte[] pattern) {logger.debug("send redis info");messageInfoUtil.sendMessageByMessageInfoByte(message.getBody());}
}

StreamMessageListener

package com.example.im.infra.executor.send.stream;import com.example.im.infra.executor.send.util.MessageInfoUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;import java.util.function.Function;/*** @author PC* 消息队列监听*/
@Component
public class StreamMessageListener {private final static Logger logger = LoggerFactory.getLogger(StreamSendExecutor.class);private MessageInfoUtil messageInfoUtil;@Autowiredpublic void setMessageInfoUtil(MessageInfoUtil messageInfoUtil) {this.messageInfoUtil = messageInfoUtil;}@Beanpublic Function<Flux<Message<byte[]>>, Mono<Void>> listener() {logger.debug("send stream info");return messageInfoFlux -> messageInfoFlux.map(message -> messageInfoUtil.sendMessageByMessageInfoByte(message.getPayload())).then();}
}

未用到的bean不加载

当未用到某些渠道时，无需进行相关配置，如使用了redis渠道，yml中就无需配置kafka信息

Redis渠道

调整config文件，当渠道非redis时对redis渠道相关bean不进行加载

WebSocketConfig

package com.example.im.config;import com.example.im.infra.handle.ImRejectExecutionHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;import javax.annotation.Resource;/*** @author PC*/
@Configuration
@EnableWebSocket
public class WebSocketConfig {@Resourceprivate WebSocketProperties webSocketProperties;@Beanpublic ServerEndpointExporter serverEndpoint() {return new ServerEndpointExporter();}/**** 配置线程池* @return 线程池*/@Beanpublic TaskExecutor taskExecutor() {WebSocketProperties.ExecutorProperties executorProperties = webSocketProperties.getExecutorProperties();ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();// 设置核心线程数executor.setCorePoolSize(executorProperties.getCorePoolSize());// 设置最大线程数executor.setMaxPoolSize(executorProperties.getMaxPoolSize());// 设置队列容量executor.setQueueCapacity(executorProperties.getQueueCapacity());// 设置线程活跃时间（秒）executor.setKeepAliveSeconds(executorProperties.getKeepAliveSeconds());// 设置默认线程名称executor.setThreadNamePrefix("im-");// 设置拒绝策略executor.setRejectedExecutionHandler(new ImRejectExecutionHandler());// 等待所有任务结束后再关闭线程池executor.setWaitForTasksToCompleteOnShutdown(true);return executor;}
}

RedisConfiguration

缓存用到了redis,RedisTemplate需要加载

package com.example.im.config;import com.example.im.infra.executor.send.redis.RedisMessageListener;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
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.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.StringRedisSerializer;/*** @author PC*/
@Configuration
public class RedisConfiguration {/*** redisTemplate配置** @return RedisTemplate*/@Beanpublic RedisTemplate<String, String> redisTemplate(RedisConnectionFactory connectionFactory) {RedisTemplate<String, String> redisTemplate = new RedisTemplate<>();StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();redisTemplate.setKeySerializer(stringRedisSerializer);redisTemplate.setValueSerializer(stringRedisSerializer);redisTemplate.setStringSerializer(stringRedisSerializer);redisTemplate.setDefaultSerializer(stringRedisSerializer);redisTemplate.setHashKeySerializer(stringRedisSerializer);redisTemplate.setHashValueSerializer(stringRedisSerializer);redisTemplate.setConnectionFactory(connectionFactory);return redisTemplate;}@Bean@ConditionalOnProperty(name = "cus.ws.communication-type", havingValue = "redis")RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,MessageListenerAdapter listenerAdapter) {RedisMessageListenerContainer container = new RedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);// 订阅一个或多个频道container.addMessageListener(listenerAdapter, new PatternTopic("redis-websocket-*"));return container;}@Bean@ConditionalOnProperty(name = "cus.ws.communication-type", havingValue = "redis")MessageListenerAdapter listenerAdapter(RedisMessageListener redisMessageListener) {return new MessageListenerAdapter(redisMessageListener);}
}

RedisMessageListener

@Component
@ConditionalOnProperty(name = "cus.ws.communication-type", havingValue = "redis")
public class RedisMessageListener implements MessageListener

RedisSendExecutor

@Component
@ConditionalOnProperty(name = "cus.ws.communication-type", havingValue = "redis")
public class RedisSendExecutor extends AbstractBaseSendExecutor

Kafka渠道

StreamMessageListener

@Component
@ConditionalOnProperty(name = "cus.ws.communication-type", havingValue = "stream")
public class StreamMessageListener

StreamSendExecutor

@Component
@ConditionalOnProperty(name = "cus.ws.communication-type", havingValue = "stream")
public class StreamSendExecutor extends AbstractBaseSendExecutor

参考资料

[1].im项目

即时通讯代码优化

在线用户逻辑修复在进行测试时，发现当前代码有个问题，如果test1在服务器进行连接，本地的test2给test1发消息，虽然test1能收到服务器上的信息，但是本地服务日志中会报teset1不在线，需要对该种情况进行修复…...

编程日记 2024/10/22 4:05:28

jmeter学习（8）界面的使用

1、新建test plan 3、打开文件 4、保存 5、剪切 6、复制 7、粘贴 8、所有线程组展开 9、所有线程组收缩 10、置灰，操作后无法使用 11、执行 13、清空当前线程组结果 14、清空所有线程组结果 15、函数助手搜索，可以用于搜索某个请求&#x…...

编程日记 2024/10/22 4:03:26

记录一次hiveserver2卡死(假死)问题

问题描述给开发人员开通了个账号，连接hive进行查询，后来发现，hive服务有时候会卡死，查询不了，连不上（所有账号/客户端都连不上hive），但在chd里面看监控，服务器资源状态…...

编程日记 2024/10/22 4:02:24

【ios】在 SwiftUI 中实现可随时调用的加载框

在 SwiftUI 项目中实现一个自定义的加载框（loading）功能，可以在任意位置调用，以便显示加载动画或者进度条。下面的教程将详细讲解如何创建一个可复用的 Loading 组件，并通过通知机制控制其显示和隐藏。先上效果&…...

编程日记 2024/10/22 4:01:24

字符、解释型语言、编程语言的互操作、输出

字符同样是1，有人看到的是数字，有人看到的是字符，还有人看到的是一个小目标。不同语言的字符正则表达式把字符分成普通字符和元字符，元字符为了搭配匹配。比如.代表任意非换行字符，这对于通配很简便，用\…...

编程日记 2024/10/22 4:00:22

基于Python的自然语言处理系列（39）：Huggingface中的解码策略

在自然语言生成任务中，如何选择下一步的单词或者词语对生成的文本质量影响巨大。Huggingface 提供了多种解码策略，可以在不同的场景下平衡流畅度、创造力以及生成效率。在这篇文章中，我们将逐步介绍 Huggingface 中的几种常见解码策略&#x…...

编程日记 2024/10/22 3:58:20

如何将视频格式转为mp4？好好看看下面这几个方法

如何将视频格式转为mp4？在数字化时代，视频已成为信息传播与娱乐消遣的重要载体。无论是学习、工作还是日常生活，我们几乎每天都会接触到各式各样的视频内容。然而，不同设备、平台或软件生成的视频文件往往采用不同的编码格式&…...

编程日记 2024/10/22 3:55:18

景区智慧公厕系统，监测公厕异味，自动清洁除臭

随着旅游业的快速发展，景区的公共厕所管理成为提升游客体验的重要环节。传统的公厕管理方式存在诸多不足，如卫生条件差、异味严重等问题。为了改善这些问题，许多景区开始采用智慧公厕系统。这种系统能够实时监测公厕内的异味，并自…...

编程日记 2024/10/22 3:54:17

GitLab CVE-2024-6389、CVE-2024-4472 漏洞解决方案

极狐GitLab 近日发布安全补丁版本17.3.2, 17.2.5, 17.1.7，修复了17个安全漏洞，本分分享其中两个漏洞 CVE-2024-6389、CVE-2024-4472 两个漏洞详情及解决方案。极狐GitLab 正式推出面向 GitLab 老旧版本免费用户的专业升级服务，为 GitLab 老…...

编程日记 2024/10/22 3:53:16

hashCode的底层原理

HashCode是计算机科学中一个广泛使用的概念，特别是在Java等编程语言中，它扮演着重要的角色。为了详细解释hashCode的底层原理，以下从几个方面进行阐述： 一、hashCode的基本概念 HashCode，即哈希码，是一个将…...

编程日记 2024/10/22 3:52:15

hadoop_hdfs详解

HDFS秒懂 HDFS定义HDFS优缺点优点缺点 HDFS组成架构NameNodeDataNodeSecondary NameNodeClient NameNode工作机制元数据的存储启动流程工作流程 Secondary NameNode工作机制checkpoint工作流程 DataNode工作机制工作流程数据完整性文件块大小块太小的缺点块太大的缺点文件写入…...

编程日记 2024/10/22 3:51:14

【Linux】Linux命令行与环境变量

1.命令行前⾯写C语⾔时，很少关注过 main 函数的参数，也没有考虑过 main 为什么会有参数。实际上在C语⾔中， main 函数⼀共有三个参数，在命令⾏部分先关注前两个参数： 1. argc：表示 main 函数接收到参…...

编程日记 2024/10/22 3:50:12

改变函数调用上下文：apply与call方法详解及实例

目录改变函数调用上下文：apply与call方法详解及实例一、什么是 apply 方法？ 1、apply 语法 2、apply 示例二、什么是 call 方法？ 1、call 语法 2、call 示例三、apply 和 call 的共同与差异 1、apply 和 call 的共同点 2、apply…...

编程日记 2024/10/22 3:49:11

k8s中的微服务

一、什么是微服务用控制器来完成集群的工作负载，那么应用如何暴漏出去？需要通过微服务暴漏出去后才能被访问 Service是一组提供相同服务的Pod对外开放的接口。借助Service，应用可以实现服务发现和负载均衡。 service默认只支持4层负载均…...

编程日记 2024/10/22 3:45:06

树莓派--AI视觉小车智能机器人--1.树莓派系统烧入及WiFi设置并进入jupyterlab

一、Raspberry Pi 系统烧入使用树莓派，我们是需要有操作系统的。默认情况下，树莓派会在插入的SD卡上查找操作系统。这需要一台电脑将存储设备映像为引导设备，并将存储设备插入该电脑。大多数树莓派用户选择microSD卡作为引导设备。 1.1 下载…...

编程日记 2024/10/22 3:44:05

文章目录一、下载地址二、下载注册机三、安装教程四、启动burpsuit五、免责声明一、下载地址 https://portswigger-cdn.net/burp/releases/download?productpro&version2024.7.1&typeMacOsx二、下载注册机 https://github.com/NepoloHebo/BurpSuite-BurpLoaderKey…...

编程日记 2024/10/22 3:41:02

【AI工具大全】《史上最全的AI工具合集》

一.AI编程类工具 1.CodeArts Snap CodeArts Snap是华为云研发的智能开发助手，覆盖软件开发全生命周期，提供代码生成、研发知识问答、智能协同等功能。通过自然语言编程，它能自动生成代码、解释代码逻辑、提供调试与检查，提升开发效率和软件质量。 2.ModelArts ModelArt…...

编程日记 2024/10/22 3:39:00