使用之前首先需要Netty的Maven包：

		io.nettynetty-all4.1.84.Final

一、服务端

服务声明类，TcpServer.class

package com.supcon.supfusion.oms.tankinfo.service.closedpath.tcp;import cn.hutool.core.collection.CollectionUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;import java.util.Map;/*** @author ZhaoXu* @date 2022/11/22 16:42*/
@Slf4j
public class TcpServer {private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();private EventLoopGroup bossGroup;private EventLoopGroup workerGroup;private ServerBootstrap server;private ChannelFuture channelFuture;private Integer port;public TcpServer(Integer port) {this.port = port;// nio连接处理池this.bossGroup = new NioEventLoopGroup();// 处理事件池this.workerGroup = new NioEventLoopGroup();server = new ServerBootstrap();server.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {// 自定义处理类ch.pipeline().addLast(new TcpServerHandler());ch.pipeline().addLast(new NettyDecoder());ch.pipeline().addLast(new NettyEncoder());}});server.option(ChannelOption.SO_BACKLOG, 128);server.childOption(ChannelOption.SO_KEEPALIVE, true);}public synchronized void startListen() {try {// 绑定到指定端口channelFuture = server.bind(port).sync();log.info("netty服务器在[{}]端口启动监听", port);} catch (Exception e) {log.error("netty服务器在[{}]端口启动监听失败", port);e.printStackTrace();}}public void sendMessageToClient(String clientIp, Object msg) {Map channelMap = TcpServerHandler.channelSkipMap.get(port);Channel channel = channelMap.get(clientIp);String sendStr;try {sendStr = OBJECT_MAPPER.writeValueAsString(msg);} catch (JsonProcessingException e) {throw new RuntimeException(e);}try {log.info("向客户端 {} 发送消息内容：{}", clientIp, sendStr);channel.writeAndFlush(sendStr);} catch (Exception var4) {log.error("向客户端 {} 发送消息失败，消息内容：{}", clientIp, sendStr);throw new RuntimeException(var4);}}public void pushMessageToClients(Object msg) {Map channelMap = TcpServerHandler.channelSkipMap.get(port);if (CollectionUtil.isNotEmpty(channelMap)) {channelMap.forEach((k, v) -> sendMessageToClient(k, msg));}}
}

handler接收处理类，继承SimpleChannelInboundHandler

package com.supcon.supfusion.oms.tankinfo.service.closedpath.tcp;import cn.hutool.core.collection.CollectionUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;/*** @author ZhaoXu* @date 2022/11/22 16:43*/
@Slf4j
public class TcpServerHandler extends SimpleChannelInboundHandler