做网站看好金石网络,小程序开发平台官网,网站开发入哪个会计科目,广东省住房和城乡建设局网站目录 基于Netty实现的简单聊天服务组件效果展示技术选型#xff1a;功能分析聊天服务基础设施配置#xff08;基于Netty#xff09;定义组件基础的配置#xff08;ChatProperties#xff09;定义聊天服务类#xff08;ChatServer#xff09;定义聊天服务配置初始化类功能分析聊天服务基础设施配置基于Netty定义组件基础的配置ChatProperties定义聊天服务类ChatServer定义聊天服务配置初始化类ChatServerInitializer 用户上线、下线处理客户端绑定服务处理类ClientInboundHandler 用户消息发送、接收处理定义一个文本消息处理器TextWebSocketFrameHandler 用户登录凭证校验定义一个凭证处理器接口AuthorizationProcessor 定义 ChatAutoConfiguration 自动化配置类定义 ChatServerApplication 服务启动类 参考资料 基于Netty实现的简单聊天服务组件 本文摘自Quan后台管理服务框架中的quan-chat工具该工具仅实现了非常简单服务模型。后期本人会视情况扩展更多复杂的业务场景。 如果本文对您解决问题有帮助欢迎到Gitee或Github点个star quan-chat 是一个基于 Netty 实现的服务端即时消息通讯组件组件本身不具备业务处理能力主要的作用是提供服务端消息中转 通过实现组件中的接口可以完成与项目相关的业务功能 例如点对点消息收发、权限校验、聊天记录保存等。
web展示层ui基于layim。layim展示的功能较为丰富。为演示服务组件仅实现点对点聊天功能。其它功能视情况扩展。
本组件仅用于学习交流使用本文应用到的 layim 来自互联网如果您想将 layim 框架用于其它用途必须取得原作者授权: layui 否则产生的一切法律责任与本作者无关。
效果展示 技术选型 spring-boot-2.7.16 netty-4.1.97 layim-3.9.8 功能分析
聊天服务基础设施配置基于Netty用户上线、下线处理用户消息发送、接收处理用户登录凭证校验 完整的组件代码开源地址https://gitee.com/quan100/quan/tree/main/quan-tools/quan-chat 下面仅展示部分代码 聊天服务基础设施配置基于Netty
Netty 是一个异步事件驱动的网络应用框架用于快速开发可维护的高性能服务器和客户端。
定义组件基础的配置ChatProperties ChatProperties 主要用于定义组件内部使用到的配置参数。 package cn.javaquan.tools.chat.autoconfigure;import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.Assert;/*** Configuration properties for im support.** author javaquan* since 1.0.0*/
ConfigurationProperties(prefix quan.im)
public class ChatProperties {/*** 默认数据包最大长度* 64kb*/private final static int MAX_FRAME_SIZE 65536;/*** 默认的消息体最大长度* 64kb*/private final static int MAX_CONTENT_LENGTH 65536;/*** 空闲检查时间单位秒*/private final static long READER_IDLE_TIME 600L;/*** 开启IM服务的端口*/private Integer port;/*** SSL配置*/private Ssl ssl;/*** websocket 路径*/private String websocketPath;/*** 数据包最大长度* 单位字节*/private Integer maxFrameSize;/*** 消息体最大长度* 单位字节*/private Integer maxContentLength;/*** 允许连接空闲的最大时间* p* 当空闲超过最大时间后强制下线*/private Long readerIdleTime;public Integer getPort() {return port;}public void setPort(Integer port) {this.port port;}public int determineDefaultPort() {Assert.notNull(this.port, [Assertion failed chat server port] - this numeric argument must have value; it must not be null);return this.port;}public Ssl getSsl() {return ssl;}public void setSsl(Ssl ssl) {this.ssl ssl;}public String getWebsocketPath() {return websocketPath;}public void setWebsocketPath(String websocketPath) {this.websocketPath websocketPath;}public String determineDefaultWebsocketPath() {Assert.hasText(this.websocketPath, [Assertion failed chat server websocketPath] - it must not be null or empty);return this.websocketPath;}public Integer getMaxFrameSize() {return maxFrameSize;}public void setMaxFrameSize(Integer maxFrameSize) {this.maxFrameSize maxFrameSize;}public Integer determineDefaultMaxFrameSize() {if (null maxFrameSize) {this.setMaxFrameSize(MAX_FRAME_SIZE);}return this.maxFrameSize;}public Integer getMaxContentLength() {return maxContentLength;}public void setMaxContentLength(Integer maxContentLength) {this.maxContentLength maxContentLength;}public Integer determineDefaultMaxContentLength() {if (null maxContentLength) {this.setMaxContentLength(MAX_CONTENT_LENGTH);}return this.maxContentLength;}public Long getReaderIdleTime() {return readerIdleTime;}public void setReaderIdleTime(Long readerIdleTime) {this.readerIdleTime readerIdleTime;}public Long determineDefaultReaderIdleTime() {if (null readerIdleTime) {this.setReaderIdleTime(READER_IDLE_TIME);}return this.readerIdleTime;}/*** ssl properties.*/public static class Ssl {private boolean enabled false;private String protocol TLS;/*** an X.509 certificate chain file in PEM format*/private String keyCertChainFilePath;/*** a PKCS#8 private key file in PEM format*/private String keyFilePath;public boolean isEnabled() {return enabled;}public void setEnabled(boolean enabled) {this.enabled enabled;}public String getProtocol() {return protocol;}public void setProtocol(String protocol) {this.protocol protocol;}public String getKeyCertChainFilePath() {return keyCertChainFilePath;}public void setKeyCertChainFilePath(String keyCertChainFilePath) {this.keyCertChainFilePath keyCertChainFilePath;}public String determineDefaultKeyCertChainFilePath() {Assert.hasText(this.keyCertChainFilePath, [Assertion failed chat server keyCertChainFilePath] - it must not be null or empty);return this.keyCertChainFilePath;}public String getKeyFilePath() {return keyFilePath;}public void setKeyFilePath(String keyFilePath) {this.keyFilePath keyFilePath;}public String determineDefaultKeyFilePath() {Assert.hasText(this.keyFilePath, [Assertion failed chat server keyFilePath] - it must not be null or empty);return this.keyFilePath;}}public void afterPropertiesSet() {determineDefaultPort();determineDefaultWebsocketPath();determineDefaultMaxFrameSize();determineDefaultMaxContentLength();determineDefaultReaderIdleTime();}
}
yml 配置示例
quan: im:port: 10000 # 配置chat服务端口websocket-path: /chat # 配置chat服务websocket访问的urireader-idle-time: 1800 #允许连接空闲的时间单位秒。超时后强制下线定义聊天服务类ChatServer 用于实现客户端与服务器建立连接状态维护 package cn.javaquan.tools.chat.server;import cn.javaquan.tools.chat.autoconfigure.ChatProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.ImmediateEventExecutor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.util.Assert;import java.net.InetSocketAddress;/*** 默认的聊天服务** author javaquan* since 1.0.0*/
public class ChatServer {private static final Log logger LogFactory.getLog(ChatServer.class);private final ChannelGroup channelGroup new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);private final EventLoopGroup group new NioEventLoopGroup();private Channel channel;public ChannelFuture start(InetSocketAddress address, ChatProperties properties) {ServerBootstrap bootstrap new ServerBootstrap();bootstrap.group(group).channel(NioServerSocketChannel.class).childHandler(createInitializer(channelGroup, properties));ChannelFuture future bootstrap.bind(address);future.syncUninterruptibly();channel future.channel();return future;}protected ChannelInitializerChannel createInitializer(ChannelGroup group, ChatProperties properties) {return new ChatServerInitializer(group, properties);}public void destroy() {if (channel ! null) {channel.close();}channelGroup.close();group.shutdownGracefully();}public void start(ChatProperties properties) {ChannelFuture future this.start(new InetSocketAddress(properties.getPort()), properties);addShutdownHook(this);future.addListener((listener) - {Assert.isTrue(listener.isSuccess(), logMessageFormat(properties.getPort(), error));logger.info(logMessageFormat(properties.getPort(), success));});}/*** Registers a new virtual-machine shutdown hook.** param chatServer*/private void addShutdownHook(ChatServer chatServer) {Runtime.getRuntime().addShutdownHook(new Thread(chatServer::destroy));}private String logMessageFormat(Integer port, String state) {return String.format(%s started %s on port(s): %s, this.getClass().getSimpleName(), state, port);}
}
定义聊天服务配置初始化类ChatServerInitializer 主要用于初始化聊天服务应用到的处理器。 package cn.javaquan.tools.chat.server;import cn.javaquan.tools.chat.autoconfigure.ChatProperties;
import cn.javaquan.tools.chat.context.ClientInboundHandler;
import cn.javaquan.tools.chat.context.TextWebSocketFrameHandler;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;import java.util.concurrent.TimeUnit;/*** 初始化服务配置** author javaquan*/
public class ChatServerInitializer extends ChannelInitializerChannel {private final ChannelGroup group;private final ChatProperties properties;public ChatServerInitializer(ChannelGroup group, ChatProperties properties) {this.group group;this.properties properties;}Overrideprotected void initChannel(Channel ch) throws Exception {ChannelPipeline pipeline ch.pipeline();pipeline.addLast(new HttpServerCodec());pipeline.addLast(new ChunkedWriteHandler());pipeline.addLast(new HttpObjectAggregator(properties.getMaxContentLength()));pipeline.addLast(new IdleStateHandler(properties.getReaderIdleTime(), 0, 0, TimeUnit.SECONDS));pipeline.addLast(new ClientInboundHandler(group, properties.getWebsocketPath()));pipeline.addLast(new TextWebSocketFrameHandler());pipeline.addLast(new WebSocketServerProtocolHandler(properties.getWebsocketPath(), null, true, properties.getMaxFrameSize()));}
}
用户上线、下线处理
客户端绑定服务处理类ClientInboundHandler 主要用于处理用户上线、下线状态处理。 package cn.javaquan.tools.chat.context;import cn.javaquan.tools.chat.core.ChannelPool;
import cn.javaquan.tools.chat.core.support.AuthorizationProcessor;
import cn.javaquan.tools.chat.util.SpringUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;/*** 客户端用户状态处理** author javaquan*/
Sharable
public class ClientInboundHandler extends ChannelInboundHandlerAdapter {private static final Log logger LogFactory.getLog(ClientInboundHandler.class);private final ChannelGroup group;private final String websocketPath;public ClientInboundHandler(ChannelGroup group, String websocketPath) {this.group group;this.websocketPath websocketPath;}Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof FullHttpRequest) {FullHttpRequest request (FullHttpRequest) msg;String uri request.uri();MapString, String queryParams paramsParser(uri);online(ctx.channel(), queryParams);request.setUri(websocketPath);}super.channelRead(ctx, msg);}Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {if (evt instanceof IdleStateEvent) {IdleStateEvent event (IdleStateEvent) evt;if (event.state() IdleState.READER_IDLE) {logger.info(String.format(用户[%s]闲置时间超过最大值将关闭连接, ChannelPool.getSessionState(ctx.channel())));ctx.channel().close();}} else {super.userEventTriggered(ctx, evt);}}Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {group.add(ctx.channel());}Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {Channel channel ctx.channel();group.remove(channel);offline(channel);}/*** 异常时调用** param ctx* param cause*/Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {logger.error(服务器错误, cause);offline(ctx.channel());// 发生异常之后关闭连接关闭channelctx.channel().close();}/*** url参数解析** param uriParams* return* throws URISyntaxException*/private MapString, String paramsParser(String uriParams) throws URISyntaxException {URI uri new URI(uriParams);MapString, String paramsMap new HashMap();String queryParam uri.getQuery();String[] queryParams queryParam.split();for (String param : queryParams) {String[] urlParam param.split();paramsMap.put(urlParam[0], urlParam[1]);}return paramsMap;}/*** 用户上线** param channel* param urlParams url参数*/private void online(Channel channel, MapString, String urlParams) {String userId urlParams.get(userId);String authorization urlParams.get(authorization);AuthorizationProcessor authorizationProcessor SpringUtils.getBean(AuthorizationProcessor.class);if (!authorizationProcessor.checkAuth(authorization)) {channel.close();logger.info(String.format(用户[%s]凭证校验失败连接被服务器拒绝, userId));return;}logger.info(String.format(用户[%s]上线, userId));channel.attr(ChannelPool.SESSION_STATE).set(userId);ChannelPool.addChannel(userId, channel);/// TODO 若用户上线则通知好友已上线。kafka发送上线事件}/*** 用户离线** param channel*/private void offline(Channel channel) {ChannelPool.removeChannel(channel);logger.info(String.format(用户[%s]下线, ChannelPool.getSessionState(channel)));/// TODO 若用户下线则通知好友已下线。kafka发送下线事件}}
用户消息发送、接收处理
定义一个文本消息处理器TextWebSocketFrameHandler 用于将用户发送的文本消息转换为服务端使用的模版消息。 通过模版将消息转发给接收者。 package cn.javaquan.tools.chat.context;import cn.javaquan.tools.chat.core.MessageHandlerFactory;
import cn.javaquan.tools.chat.core.message.MessageTemplate;
import cn.javaquan.tools.chat.util.JsonUtils;
import cn.javaquan.tools.chat.util.SpringUtils;
import cn.javaquan.tools.chat.core.support.IMessageHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;/*** 消息处理器** author javaquan*/
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandlerTextWebSocketFrame {Overridepublic void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {MessageTemplate messageTemplate messageConvertor(msg);messageHandler(ctx, messageTemplate);}/*** 消息处理* p* 根据消息类型处理消息* p* 需要自定义实现{link IMessageHandler}接口。** param ctx* param messageTemplate*/private void messageHandler(ChannelHandlerContext ctx, MessageTemplate messageTemplate) {MessageHandlerFactory messageHandlerFactory SpringUtils.getBean(MessageHandlerFactory.class);messageHandlerFactory.getService(messageTemplate.getType()).handler(ctx, messageTemplate);}/*** 将字符串信息转换为模版信息格式** param msg* return*/private MessageTemplate messageConvertor(TextWebSocketFrame msg) {return JsonUtils.parseObject(msg.text(), MessageTemplate.class);}
}
用户登录凭证校验
定义一个凭证处理器接口AuthorizationProcessor 将处理器定义成接口主要目的是将组件与业务解耦。 因为不同的业务实现的权限业务都可能不一样。 只需业务端实现该接口当权限校验不通过时组件内部就会拒绝客户端连接。 package cn.javaquan.tools.chat.core.support;/*** 授权凭证处理器** author wangquan*/
public interface AuthorizationProcessor {/*** 检查权限** param authorization 登录凭证* return*/boolean checkAuth(String authorization);}
定义 ChatAutoConfiguration 自动化配置类 ChatAutoConfiguration 是 quan-chat 组件中最重要的一项配置通过该配置来定义组件是否生效。 当引入 quan-chat 组件时不需要对组件进行扫描。服务启动时会自动发现该配置。 通过该配置初始化聊天服务所依赖的相关功能。若未按照配置要求配置属性quan-chat 组件引入将无效。 package cn.javaquan.tools.chat.autoconfigure;import cn.javaquan.tools.chat.ChatServerApplication;
import cn.javaquan.tools.chat.core.ChannelPool;
import cn.javaquan.tools.chat.core.support.AbstractAuthorizationCheckProcessor;
import cn.javaquan.tools.chat.core.support.AuthorizationProcessor;
import cn.javaquan.tools.chat.server.ChatServer;
import cn.javaquan.tools.chat.server.SecureChatServer;
import io.netty.channel.Channel;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.condition.AnyNestedCondition;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;import java.io.File;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** im聊天sdk配置** author javaquan* since 1.0.0*/
AutoConfiguration
EnableConfigurationProperties(ChatProperties.class)
public class ChatAutoConfiguration {Import(ChatServerApplication.class)Configuration(proxyBeanMethods false)Conditional(ChatCondition.class)protected static class ChatConfiguration {ConditionalOnProperty(prefix quan.im.ssl, name enabled, havingValue false, matchIfMissing true)ConditionalOnMissingBeanBeanChatServer chatServer() {return new ChatServer();}ConditionalOnMissingBeanBeanChannelPool channelPool() {MapString, Channel channelContainer new ConcurrentHashMap();return new ChannelPool(channelContainer);}ConditionalOnMissingBeanBeanAuthorizationProcessor authorizationProcessor() {return new AbstractAuthorizationCheckProcessor();}}static class ChatCondition extends AnyNestedCondition {ChatCondition() {super(ConfigurationPhase.PARSE_CONFIGURATION);}ConditionalOnProperty(prefix quan.im, name port)static class PortProperty {}ConditionalOnProperty(prefix quan.im.ssl, name enabled, havingValue true)ConditionalOnMissingBeanBeanSslContext sslContext(ChatProperties properties) throws Exception {ChatProperties.Ssl ssl properties.getSsl();File keyCertChainFile new File(ssl.determineDefaultKeyCertChainFilePath());File keyFile new File(ssl.determineDefaultKeyFilePath());return SslContextBuilder.forServer(keyCertChainFile, keyFile).build();}ConditionalOnProperty(prefix quan.im.ssl, name enabled, havingValue true)ConditionalOnMissingBeanBeanChatServer secureChatServer(SslContext context) {return new SecureChatServer(context);}}}
定义 ChatServerApplication 服务启动类 当引入 quan-chat 组件时并正确配置 ChatProperties 属性服务启动时则会自动扫描 ChatServerApplication 类用于启动 聊天服务端。 package cn.javaquan.tools.chat;import cn.javaquan.tools.chat.autoconfigure.ChatProperties;
import cn.javaquan.tools.chat.server.ChatServer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;/*** chat服务启动** author javaquan* since 1.0.0*/
public class ChatServerApplication implements ApplicationRunner {Autowiredprivate ChatServer chatServer;Autowiredprivate ChatProperties properties;Overridepublic void run(ApplicationArguments args) throws Exception {properties.afterPropertiesSet();chatServer.start(properties);}
}参考资料 如果本文对您解决问题有帮助欢迎到Gitee或Github点个star quan-chat 工具文档https://doc.javaquan.cn/pages/tools/chat/ quan-chat 工具开源地址https://gitee.com/quan100/quan/tree/main/quan-tools/quan-chat
