系列文章目录

Springboot项目集成Netty组件
Netty新增解析数据包指定长度功能

---

---

前言

项目中有时候会需与其他客户端或者系统建立Sokcet连接 或者WebSocket连接,满足业务需求。Netty 是一个利用 Java 的高级网络的能力，隐藏其背后的复杂性而提供一个易于使用的 API 的客户端/服务器框架。我们可以便捷的利用此组件搭建自己的客户端/服务器框架，进行二次开发，满足自己的业务需求。

本文介绍了Springboot集成Netty的配置以及使用方式，项目是基于ruoyi分离版作为基础项目进行二次开发，如果需要完整项目，可以私聊我，如果有时间会整理出来。

---

一、Netty是什么？

Netty是由JBOSS提供的一个java开源框架，现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。
也就是说，Netty 是一个基于NIO的客户、服务器端编程框架，使用Netty 可以确保你快速和简单的开发出一个网络应用，例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程，例如：基于TCP和UDP的socket服务开发。
“快速”和“简单”并不用产生维护性或性能上的问题。Netty 是一个吸收了多种协议（包括FTP、SMTP、HTTP等各种二进制文本协议）的实现经验，并经过相当精心设计的项目。最终，Netty 成功的找到了一种方式，在保证易于开发的同时还保证了其应用的性能，稳定性和伸缩性。

简单说： Netty是一个高性能、高可靠性的基于NIO封装的网络应用框架

二、使用步骤

1. 项目引入依赖

1.1 项目基础版本信息：

Java 版本1.8
SpringBoot 版本 2.2.13.RELEASE

1.2 Netty依赖

        <dependency><groupId>io.netty</groupId><artifactId>netty-all</artifactId><version>4.1.68.Final</version></dependency>

2. 项目配置

2.1 在 yml 配置文件中配置以下：

# netty-server 配置
netty:# 监听端口port: 8180# 队列池 - 最大队列数 200queue-max: 200# 线程队列容量 10pool-queue-init: 10# netty 请求路径path: /ws

2.2 创建Netty配置类

@Data
@Component
//application.yml此文件在ruoyi-admin模块中
@PropertySource(name="NettyProperties",value = {"classpath:application.yml"},ignoreResourceNotFound=false, encoding="UTF-8")
@ConfigurationProperties(prefix = "netty")
public class NettyProperties {private Integer port;private Integer queueMax;private String path;
}

3. Netty 服务端

3.1 Netty 服务端

import com.ruoyi.entrust.constant.NettyProperties;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2021/12/26 18:18*/
@Slf4j
@Component
public class NettySocketServer {@Autowiredprivate NettyProperties properties;public void start() throws Exception {// Boss线程：由这个线程池提供的线程是boss种类的，用于创建、连接、绑定socket， （有点像门卫）然后把这些socket传给worker线程池。// 在服务器端每个监听的socket都有一个boss线程来处理。在客户端，只有一个boss线程来处理所有的socket。EventLoopGroup bossGroup = new NioEventLoopGroup();// Worker线程：Worker线程执行所有的异步I/O，即处理操作EventLoopGroup workgroup = new NioEventLoopGroup();try {// ServerBootstrap 启动NIO服务的辅助启动类,负责初始话netty服务器，并且开始监听端口的socket请求ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workgroup)// 设置非阻塞,用它来建立新accept的连接,用于构造serversocketchannel的工厂类.channel(NioServerSocketChannel.class)// ServerChannelInitializer 对出入的数据进行的业务操作,其继承ChannelInitializer.childHandler(new ServerChannelInitializer())//设置队列大小.option(ChannelOption.SO_BACKLOG, properties.getQueueMax())// 两小时内没有数据的通信时,TCP会自动发送一个活动探测数据报文.childOption(ChannelOption.SO_KEEPALIVE, true);ChannelFuture channelFuture = serverBootstrap.bind(properties.getPort()).sync();log.info("Netty服务器开启等待客户端连接, 监听端口:{}", properties.getPort());channelFuture.channel().closeFuture().sync();} catch (Exception e) {e.printStackTrace();} finally {// 关闭主线程组bossGroup.shutdownGracefully();// 关闭工作线程workgroup.shutdownGracefully();}}
}

3.2 Netty channle注册类

对出入的数据进行的业务操作进行初始化

目前通道支持连接协议 WebSocket和 Socket

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2021/12/26 18:32** channle注册类*/
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch)  {ChannelPipeline pipeline = ch.pipeline();//添加日志pipeline.addLast("logging",new LoggingHandler(LogLevel.WARN));// 判断 连接类型 此处 用来支持多种连接 目前支持 WebSocket 和 Socketpipeline.addLast("socketChoose",new SocketChooseHandler());}
}

3.3 Netty 协议初始化解码器

此类用于 解析多种协议，调用对应协议的处理器

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketFrameAggregator;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;import java.nio.ByteOrder;
import java.util.List;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2022/1/18 10:34*** 协议初始化解码器.* 用来判定实际使用什么协议.*/
@Slf4j
public class SocketChooseHandler extends ByteToMessageDecoder {/** 默认暗号长度为23 */private static final int MAX_LENGTH = 23;/** WebSocket握手的协议前缀 */private static final String WEBSOCKET_PREFIX = "GET /";@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {String protocol = getBufStart(in);in.resetReaderIndex();// 输出数据log.debug("获取的数据为: {}",protocol);if (protocol.startsWith(WEBSOCKET_PREFIX)) {log.info("解析 webSocket");//  websocket连接时，执行以下处理// HttpServerCodec：将请求和应答消息解码为HTTP消息ctx.pipeline().addLast("http-codec", new HttpServerCodec());// HttpObjectAggregator：将HTTP消息的多个部分合成一条完整的HTTP消息ctx.pipeline().addLast("aggregator", new HttpObjectAggregator(65535));// ChunkedWriteHandler：向客户端发送HTML5文件,文件过大会将内存撑爆ctx.pipeline().addLast("http-chunked", new ChunkedWriteHandler());ctx.pipeline().addLast("WebSocketAggregator", new WebSocketFrameAggregator(65535));//  若超过6000秒未收到约定心跳，则主动断开channel释放资源ctx.pipeline().addLast(new IdleStateHandler(6000,0,0));// 用于处理websocket URI参数ctx.pipeline().addLast(new WebSocketURIHandler());//用于处理websocket, /ws为访问websocket时的urictx.pipeline().addLast("ProtocolHandler", new WebSocketServerProtocolHandler("/ws"));ctx.pipeline().addLast(new WebSocketServerHandler());} else {log.info("解析 socket");//  常规TCP连接时，执行以下处理ctx.pipeline().addLast(new IdleStateHandler(60,0,0));// 数据读取使用小端模式ctx.pipeline().addLast("length-decoder",new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN,Integer.MAX_VALUE, 0, 4, 0, 4,true));ctx.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));ctx.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));ctx.pipeline().addLast(new SocketServerHandler());}ctx.pipeline().remove(this.getClass());}private String getBufStart(ByteBuf in){int length = in.readableBytes();if (length > MAX_LENGTH) {length = MAX_LENGTH;}// 标记读位置in.markReaderIndex();byte[] content = new byte[length];in.readBytes(content);return new String(content);}
}

在处理 Socket协议的处理中，代码中使用了小端处理方案，具体原因可见我的另外一篇博客Netty新增解析数据包指定长度功能

3.4 Netty 全局通道池

由于作为Netty 服务器端，实际业务中会创建很多通道，有可能查找指定已有通道并进行数据交互。因此，开发此全局通道池，业务中可以进行保存通道，复用，以及查询通道进行交互等功能。

以下注释是 本人实际项目中的注释，理解后，根据实际项目进行对应修改，但是此类是通用的

import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.springframework.util.StringUtils;
import java.util.List;
import java.util.stream.Collectors;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2022/1/15 16:07*/
public class GlobalChannelPool {// 临时存放连接客户端public static ChannelGroup temporaryChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);// 存放已绑定用户信息连接客户端public static ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);/*** 在集合中获取对应的通道** @param module  模块名 exam 和 entrust  可根据实际业务进行修改* @param protocol 连接类型 webSocket 和 socket  可根据实际业务进行修改* @param keyName  数据类型  exam模块key是mac | entrust模块key是uid   可根据实际业务进行修改* @param key 数据类型值 可根据实际业务进行修改* @return*/public static List<Channel> getChannelByKey(String module, String protocol ,String keyName, String key){if (StringUtils.isEmpty(module) && StringUtils.isEmpty(protocol) && StringUtils.isEmpty(keyName) && StringUtils.isEmpty(key)){return null;}List<Channel> channelList = null;String moduleName = "module";String protocolName = "protocol";try{// 先过滤 模块  examination | detection | entrustAttributeKey<String> moduleKey = AttributeKey.valueOf(moduleName);channelList = channels.stream().filter(channel -> channel.attr(moduleKey).get().equals(module)).collect(Collectors.toList());// 过滤 socket类型  webSocket 和 socketAttributeKey<String> protocolKey = AttributeKey.valueOf(protocolName);channelList = channelList.stream().filter(channel -> channel.attr(protocolKey).get().equals(protocol)).collect(Collectors.toList());// 过滤 数据类型  examination模块 身份证读卡器 key是mac | detection模块 身份证读卡器 key是mac | entrust模块 签名功能 key是signatureAttributeKey<String> dataKey = AttributeKey.valueOf(keyName);channelList = channelList.stream().filter(channel -> channel.attr(dataKey).get().equals(key)).collect(Collectors.toList());}catch (Exception e){return null;}return channelList;}/*** 通道添加属性 区分通道** @param channel 通道* @param module  模块名 exam 和 entrust* @param protocol 连接协议 webSocket 和 socket* @param keyName  数据类型  exam模块key是mac | entrust模块key是uid* @param key 数据类型值*/public static void addChannelAttrKey(Channel channel, String module, String protocol ,String keyName, String key){if (channel != null){// 存储 模块信息  examination | detection | entrustAttributeKey<String> moduleKey = AttributeKey.valueOf("module");channel.attr(moduleKey).set(module);// 存储 连接协议  webSocket 和 socketAttributeKey<String> protocolKey = AttributeKey.valueOf("protocol");channel.attr(protocolKey).set(protocol);// 存储 数据类型  examination模块 身份证读卡器 key是mac | detection模块 身份证读卡器 key是mac | entrust模块 签名功能 key是signatureAttributeKey<String> dataKey = AttributeKey.valueOf(keyName);channel.attr(dataKey).set(key);}}
}

3.5 WebSocket 协议通道数据处理

由于项目中，WebSocket请求链接可以携带参数，因此对于此参数需要提前处理，才可以进一步进行处理WebSocket链接中交互的数据。

3.5.1 WebSocket 处理URI参数

此函数用于处理 WebSocket中参数

import com.ruoyi.entrust.constant.NettyProperties;
import com.ruoyi.examination.utils.SpringContextUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2022/1/19 14:53*/
@Slf4j
public class WebSocketURIHandler extends SimpleChannelInboundHandler<WebSocketFrame> {private NettyProperties nettyProperties;public WebSocketURIHandler(){nettyProperties = (NettyProperties) SpringContextUtil.getBean("nettyProperties");}/*** 建立连接** channel 通道 action 活跃的 当客户端主动链接服务端的链接后，这个通道就是活跃的了。* 也就是客户端与服务端建立了通信通道并且可以传输数据*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 把新连接保存到临时连接表中GlobalChannelPool.temporaryChannels.add(ctx.channel());log.info("客户端与服务端连接开启：{},客户端名字：{}", ctx.channel().remoteAddress().toString(),ctx.name());}/*** 断开连接** channel 通道 Inactive 不活跃的 当客户端主动断开服务端的链接后，这个通道就是不活跃的。* 也就是说客户端与服务端关闭了通信通道并且不可以传输数据*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 从连接列表中移除该连接GlobalChannelPool.channels.remove(ctx.channel());log.info("客户端与服务端连接关闭：{},客户端名字：{}", ctx.channel().remoteAddress().toString(),ctx.name());}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (msg instanceof FullHttpRequest) {FullHttpRequest request = (FullHttpRequest) msg;String uri = request.uri();String origin = request.headers().get("Origin");log.info("uri:{},Origin:{}",uri,origin);Map<String, String> params = NettyUtil.getParams(uri);if (nettyProperties.getPath().equals(NettyUtil.getBasePath(uri))) {// 因为有可能携带了参数，导致客户端一直无法返回握手包，因此在校验通过后，重置请求路径Channel channel = ctx.channel();GlobalChannelPool.channels.add(channel);// 添加 AttributeKeyGlobalChannelPool.addChannelAttrKey(channel,params.get("module"),"webSocket",params.get("keyName"),params.get("key"));// 连接已绑定完成,把该链接移除临时存储区,避免积累过多,检索慢GlobalChannelPool.temporaryChannels.remove(channel);request.setUri(nettyProperties.getPath());log.info("webSocket处理完成，重新生成路径!!!");}else {log.warn("webSocket连接不正确。请求链接URI是：{}",uri);ctx.close();}}super.channelRead(ctx, msg);}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, WebSocketFrame msg) throws Exception {ctx.channel().writeAndFlush(msg.retain());}
}

3.5.2 WebSocket 数据处理类

package com.ruoyi.entrust.utils.netty;import io.netty.channel.*;
import io.netty.handler.codec.http.websocketx.*;
import lombok.extern.slf4j.Slf4j;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2022/1/18 14:41*/@Slf4j
@ChannelHandler.Sharable
public class WebSocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {/*** 建立连接** channel 通道 action 活跃的 当客户端主动链接服务端的链接后，这个通道就是活跃的了。* 也就是客户端与服务端建立了通信通道并且可以传输数据*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 把新连接保存到临时连接表中GlobalChannelPool.temporaryChannels.add(ctx.channel());log.info("客户端与服务端连接开启：{},客户端名字：{}", ctx.channel().remoteAddress().toString(),ctx.name());}/*** 断开连接** channel 通道 Inactive 不活跃的 当客户端主动断开服务端的链接后，这个通道就是不活跃的。* 也就是说客户端与服务端关闭了通信通道并且不可以传输数据*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 从连接列表中移除该连接GlobalChannelPool.channels.remove(ctx.channel());log.info("客户端与服务端连接关闭：{},客户端名字：{}", ctx.channel().remoteAddress().toString(),ctx.name());}/*** 接收客户端发送的消息 channel 通道 Read 读 简而言之就是从通道中读取数据，* 也就是服务端接收客户端发来的数据。但是这个数据在不进行解码时它是ByteBuf类型的*/@Overrideprotected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {log.info("接收到来自客户端的信息：{}", ctx.channel().remoteAddress().toString());if (msg != null) {// WebSocket消息处理String webSocketInfo = ((TextWebSocketFrame) msg).text().trim();log.info("收到webSocket消息：" + webSocketInfo);}else {  //数据处理为空log.warn("接收到来自客户端的信息：{}   warn：消息为空", ctx.channel().remoteAddress().toString());}}/**** channel 通道 Read 读取 Complete 完成 在通道读取完成后会在这个方法里通知，* 对应可以做刷新操作 ctx.flush()*/@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}/*** exception 异常 Caught 抓住 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {log.error("发生异常的连接是" + ctx.channel().id());log.error("netty服务端发送异常" + cause);cause.printStackTrace();ctx.close();}
}

3.6 Socket 协议通道数据处理

package com.ruoyi.entrust.utils.netty;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.constant.HttpStatus;
import com.ruoyi.common.core.domain.AjaxResult;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.entrust.constant.NettyDataParam;
import com.ruoyi.entrust.utils.JsonUtil;
import com.ruoyi.examination.domain.DetectionOrder;
import com.ruoyi.examination.domain.Order;
import com.ruoyi.examination.service.impl.DetectionOrderServiceImpl;
import com.ruoyi.examination.service.impl.ExaminationOrderServiceImpl;
import com.ruoyi.examination.utils.SpringContextUtil;
import io.netty.channel.*;
import com.ruoyi.examination.domain.Message;
import io.netty.handler.codec.http.websocketx.*;
import lombok.extern.slf4j.Slf4j;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2022/1/17 21:09*/@Slf4j
@ChannelHandler.Sharable
public class SocketServerHandler extends SimpleChannelInboundHandler<Object> {/*** 建立连接** channel 通道 action 活跃的 当客户端主动链接服务端的链接后，这个通道就是活跃的了。* 也就是客户端与服务端建立了通信通道并且可以传输数据*/@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 把新连接保存到临时连接表中GlobalChannelPool.temporaryChannels.add(ctx.channel());log.info("客户端与服务端连接开启：{},客户端名字：{}", ctx.channel().remoteAddress().toString(),ctx.name());}/*** 断开连接** channel 通道 Inactive 不活跃的 当客户端主动断开服务端的链接后，这个通道就是不活跃的。* 也就是说客户端与服务端关闭了通信通道并且不可以传输数据*/@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {// 从连接列表中移除该连接GlobalChannelPool.channels.remove(ctx.channel());log.info("客户端与服务端连接关闭：{},客户端名字：{}", ctx.channel().remoteAddress().toString(),ctx.name());}/*** 接收客户端发送的消息 channel 通道 Read 读 简而言之就是从通道中读取数据，* 也就是服务端接收客户端发来的数据。但是这个数据在不进行解码时它是ByteBuf类型的*/@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {log.info("接收到来自客户端的信息：{}", ctx.channel().remoteAddress().toString());// WebSocket消息处理if (msg instanceof WebSocketFrame) {log.warn("Socket Handler 接收到WebSocket信息： 请检查！！！");}// Socket消息处理else{String requestData = (String)msg;log.info("收到socket消息："+ requestData);}}/**** channel 通道 Read 读取 Complete 完成 在通道读取完成后会在这个方法里通知，* 对应可以做刷新操作 ctx.flush()*/@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.flush();}/*** exception 异常 Caught 抓住 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接*/@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {log.error("发生异常的连接是" + ctx.channel().id());log.error("netty服务端发送异常" + cause);cause.printStackTrace();ctx.close();}
}

3.7 Netty 公共工具类

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.constant.HttpStatus;
import com.ruoyi.common.core.domain.AjaxResult;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;import java.util.HashMap;
import java.util.List;
import java.util.Map;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2022/1/18 16:59*/
@Slf4j
public class NettyUtil {/*** 将路径参数转换成Map对象，如果路径参数出现重复参数名，将以最后的参数值为准* @param uri 传入的携带参数的路径* @return*/public static Map<String, String> getParams(String uri) {Map<String, String> params = new HashMap<>(10);int idx = uri.indexOf("?");if (idx != -1) {String[] paramsArr = uri.substring(idx + 1).split("&");for (String param : paramsArr) {idx = param.indexOf("=");params.put(param.substring(0, idx), param.substring(idx + 1));}}return params;}/*** 获取URI中参数以外部分路径* @param uri* @return*/public static String getBasePath(String uri) {if (uri == null || uri.isEmpty())return null;int idx = uri.indexOf("?");if (idx == -1)return uri;return uri.substring(0, idx);}/*** 发布消息到对应 通道** @param channel 当前连接通道* @param message 所需发布消息* @param requestJsonObject 查询通道的参数* @param protocol 通道连接协议* @return 是否有对应通道*/public static boolean pushInfoToChannels(Channel channel, String message, JSONObject requestJsonObject, String protocol){boolean haveChannels = false;// 查询对应 连接通道List<Channel> channels = GlobalChannelPool.getChannelByKey(requestJsonObject.getString("module"),protocol,requestJsonObject.getString("keyName"),requestJsonObject.getString("key"));if (null == channels || channels.size() == 0){log.warn("没有发现对应{}客户端，模块为：{}，请求连接模式为：{}，KeyName为：{}，Key为：{}", protocol,requestJsonObject.getString("module"),protocol,requestJsonObject.getString("keyName"),requestJsonObject.getString("key"));channel.writeAndFlush(JSON.toJSONString(AjaxResult.error(HttpStatus.NO_WEBSOCKET,"没有发现对应webSocket客户端")));}else {log.info("发现多个对应{}客户端，数量为：{}，模块为：{}，请求连接模式为：{}，KeyName为：{}，Key为：{}", protocol,channels.size(), requestJsonObject.getString("module"), protocol, requestJsonObject.getString("keyName"),requestJsonObject.getString("key"));haveChannels = true;}// 发送数据到对应的通道if (haveChannels){for (Channel ch: channels){ch.writeAndFlush(new TextWebSocketFrame(message));}}return haveChannels;}/*** 发布消息到对应 通道** @param message 所需发布消息* @param requestJsonObject 查询通道的参数* @param protocol 通道连接协议* @return 是否有对应通道*/public static boolean pushInfoToChannels(String message, JSONObject requestJsonObject, String protocol){boolean haveChannels = false;// 查询对应 连接通道List<Channel> channels = GlobalChannelPool.getChannelByKey(requestJsonObject.getString("module"),protocol,requestJsonObject.getString("keyName"),requestJsonObject.getString("key"));if (null == channels || channels.size() == 0){log.warn("没有发现对应{}客户端，模块为：{}，请求连接模式为：{}，KeyName为：{}，Key为：{}", protocol,requestJsonObject.getString("module"),protocol,requestJsonObject.getString("keyName"),requestJsonObject.getString("key"));}else {log.info("发现多个对应{}客户端，数量为：{}，模块为：{}，请求连接模式为：{}，KeyName为：{}，Key为：{}", protocol,channels.size(), requestJsonObject.getString("module"), protocol, requestJsonObject.getString("keyName"),requestJsonObject.getString("key"));haveChannels = true;}// 发送数据到对应的通道if (haveChannels){for (Channel ch: channels){ch.writeAndFlush(new TextWebSocketFrame(message));}}return haveChannels;}
}

3.8 Netty 启动类

此处 Netty作为单独服务器启动，与SpringBoot使用端口并不一致

通过实现CommandLineRunner接口和添加@Component注解，使启动Netty服务器在项目启动之后执行

Spring boot的CommandLineRunner接口主要用于实现在应用初始化后，去执行一段代码块逻辑，这段初始化代码在整个应用生命周期内只会执行一次。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;/*** @program: ruoyi* @Author： LiuZedi* @Date： 2022/1/15 14:55*/
@Order(1)
@Component
public class NettyStart implements CommandLineRunner {@Autowiredprivate NettySocketServer nettySocketServer;@Overridepublic void run(String... args) throws Exception {nettySocketServer.start();}
}

---

总结

以上就是今天要讲的内容，本文介绍了Netty的使用，Netty提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。我们可以快速使用此类方法进行使用开发自己的网络服务器。

之后我将写一篇文章，介绍如何使用Netty 实现 通过微信小程序 在网页上进行电子签字的功能。