SpringBoot项目整合WebSocket+netty实现前后端双向通信（同时支持前端webSocket和socket协议哦）

目录

前言

技术栈

功能展示

springboot%E9%A1%B9%E7%9B%AE%E6%B7%BB%E5%8A%A0***ty%E4%BE%9D%E8%B5%96-toc" style="margin-left:40px;">一、springboot项目添加***ty依赖

二、***ty服务端

三、***ty客户端

四、测试

五、代码仓库地址

  专属小彩蛋：前些天发现了一个巨牛的人工智能学习网站，通俗易懂，风趣幽默，忍不住分享一下给大家。点击跳转到网站（前言 - 床长人工智能教程） 

前言

        最近做了一个硬件设备通信项目，需求是这样，前端使用websocket向后端进行tcp协议的通信，后端***ty服务端收到数据后，将数据发往socket客户端，客户端收到数据之后需要进行响应数据显示到前端页面供用户进行实时监控。

技术栈

        后端

• springboot 
• ***ty

        前端

• 前端websocket

功能展示

前端页面输入webSocket地址，点击连接，输入待发送的数据，点击发送

 后端我们可以使用网络测试工具***Assist 进行响应测试

 在工具中连接***ty服务端，并点击发送按钮，可以看到，前端页面右侧对话框成功显示出了***Assist测试工具响应的数据内容。接下来我们来看一看代码如何进行实现，关键的点在于需要同时支持前端websocket和后端socket的连接，需要自定义一个协议选择处理器。

一、springboot项目添加***ty依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.12</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>***.example.dzx.***ty</groupId>
    <artifactId>qiyan-project</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>qiyan-project</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-autoconfigure</artifactId>
            <version>2.6.7</version>
        </dependency>
        <dependency>
            <groupId>io.***ty</groupId>
            <artifactId>***ty-all</artifactId>
            <version>4.1.52.Final</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

二、***ty服务端

 （1）***ty服务启动类

package ***.example.dzx.***ty.qiyanproject.server;

import io.***ty.bootstrap.ServerBootstrap;
import io.***ty.channel.ChannelFuture;
import io.***ty.channel.ChannelOption;
import io.***ty.channel.EventLoopGroup;
import io.***ty.channel.nio.NioEventLoopGroup;
import io.***ty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.***ponent;

import java.***.I***SocketAddress;

/**
 * @author dzx
 * @ClassName:
 * @Description: ***ty服务启动类
 * @date 2023年06月30日 21:27:16
 */
@Slf4j
@***ponent
public class ***tyServer {

    public void start(I***SocketAddress address) {
        //配置服务端的NIO线程组

        /*
         * 在***ty中，事件循环组是一组线程池，用于处理网络事件，例如接收客户端连接、读写数据等操作。
         * 它由两个部分组成：bossGroup和workerGroup。
         * bossGroup 是负责接收客户端连接请求的线程池。
         * workerGroup 是负责处理客户端连接的线程池。
         * */

        EventLoopGroup bossGroup = new NioEventLoopGroup(10);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            //创建ServerBootstrap实例，boss组用于接收客户端连接请求，worker组用于处理客户端连接。
            ServerBootstrap bootstrap = new ServerBootstrap()
                    .group(bossGroup, workerGroup)  // 绑定线程池
                    .channel(NioServerSocketChannel.class)//通过TCP/IP方式进行传输
                    .childOption(ChannelOption.SO_REUSEADDR, true) //快速复用端口
                    .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000)
                    .localAddress(address)//监听服务器地址
                    .childHandler(new ***tyServerChannelInitializer())
//                    .childHandler(new ***.***p.dev.system.***ty.***tyServerChannelInitializer())
                    .childOption(ChannelOption.TCP_NODELAY, true)//子处理器处理客户端连接的请求和数据
                    .option(ChannelOption.SO_BACKLOG, 1024)  //服务端接受连接的队列长度，如果队列已满，客户端连接将被拒绝
                    .childOption(ChannelOption.SO_KEEPALIVE, true);  //保持长连接，2小时无数据激活心跳机制

            // 绑定端口，开始接收进来的连接
            ChannelFuture future = bootstrap.bind(address).sync();
            future.addListener(l -> {
                if (future.isSu***ess()) {
                    System.out.println("***ty服务启动成功");
                } else {
                    System.out.println("***ty服务启动失败");
                }
            });
            log.info("***ty服务开始监听端口: " + address.getPort());
            //关闭channel和块，直到它被关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("启动***ty服务器时出错", e);
        } finally {
            //释放资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

 （2）服务端初始化类编写，客户端与服务器端连接一旦创建，这个类中方法就会被回调，设置出站编码器和入站解码器以及各项处理器

package ***.example.dzx.***ty.qiyanproject.server;

import io.***ty.channel.ChannelInitializer;
import io.***ty.channel.ChannelPipeline;
import io.***ty.channel.socket.SocketChannel;
import io.***ty.handler.timeout.IdleStateHandler;
import org.springframework.stereotype.***ponent;


/**
 * @author dzx
 * @ClassName:
 * @Description: 服务端初始化，客户端与服务器端连接一旦创建，这个类中方法就会被回调，设置出站编码器和入站解码器以及各项处理器
 * @date 2023年06月30日 21:27:16
 */
@***ponent
public class ***tyServerChannelInitializer extends ChannelInitializer<SocketChannel> {

//    private FullHttpResponse createCorsResponseHeaders() {
//        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
//
//        // 设置允许跨域访问的响应头
//        response.headers().set(HttpHeaderNames.A***ESS_CONTROL_ALLOW_ORIGIN, "*");
//        response.headers().set(HttpHeaderNames.A***ESS_CONTROL_ALLOW_METHODS, "GET, POST, PUT, DELETE");
//        response.headers().set(HttpHeaderNames.A***ESS_CONTROL_ALLOW_HEADERS, "Content-Type, Authorization");
//        response.headers().set(HttpHeaderNames.A***ESS_CONTROL_MAX_AGE, "3600");
//
//        return response;
//    }

    @Override
    protected void initChannel(SocketChannel channel) {
        ChannelPipeline pipeline = channel.pipeline();
        pipeline.addLast("active", new ChannelActiveHandler());
        //Socket 连接心跳检测
        pipeline.addLast("idleStateHandler", new IdleStateHandler(60, 0, 0));
        pipeline.addLast("socketChoose", new SocketChooseHandler());
        pipeline.addLast("***monhandler",new ***tyServerHandler());
    }
}

（3） 编写新建连接处理器

package ***.example.dzx.***ty.qiyanproject.server;

import ***.example.dzx.***ty.qiyanproject.constants.General;
import io.***ty.channel.ChannelHandler;
import io.***ty.channel.ChannelHandlerContext;
import io.***ty.channel.ChannelId;
import io.***ty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;

import java.***.I***SocketAddress;

/**
 * @author dzx
 * @ClassName:
 * @Description: 客户端新建连接处理器
 * @date 2023年06月30日 21:27:16
 */

@ChannelHandler.Sharable
@Slf4j
public class ChannelActiveHandler extends ChannelInboundHandlerAdapter {

    /**
     * 有客户端连接服务器会触发此函数
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        //获取客户端连接的远程地址
        I***SocketAddress insocket = (I***SocketAddress) ctx.channel().remoteAddress();
        //获取客户端的IP地址
        String clientIp = insocket.getAddress().getHostAddress();
        //获取客户端的端口号
        int clientPort = insocket.getPort();
        //获取连接通道唯一标识
        ChannelId channelId = ctx.channel().id();
        //如果map中不包含此连接，就保存连接
        if (General.CHANNEL_MAP.containsKey(channelId)) {
            log.info("Socket------客户端【" + channelId + "】是连接状态，连接通道数量: " + General.CHANNEL_MAP.size());
        } else {
            //保存连接
            General.CHANNEL_MAP.put(channelId, ctx);
            log.info("Socket------客户端【" + channelId + "】连接***ty服务器[IP:" + clientIp + "--->PORT:" + clientPort + "]");
            log.info("Socket------连接通道数量: " + General.CHANNEL_MAP.size());
        }
    }
}

（4）编写协议初始化解码器,用来判定实际使用什么协议(实现websocket和socket同时支持的关键点就在这里)

package ***.example.dzx.***ty.qiyanproject.server;

/**
 * @author 500007
 * @ClassName:
 * @Description:
 * @date 2023年06月30日 21:29:17
 */

import ***.example.dzx.***ty.qiyanproject.constants.General;
import io.***ty.buffer.ByteBuf;
import io.***ty.channel.ChannelHandlerContext;
import io.***ty.channel.ChannelId;
import io.***ty.handler.codec.ByteToMessageDecoder;
import io.***ty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.***ponent;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author dzx
 * @ClassName:
 * @Description:  协议初始化解码器.用来判定实际使用什么协议，以用来处理前端websocket或者后端***ty客户端的连接或通信
 * @date 2023年06月30日 21:31:24
 */
@***ponent
@Slf4j
public class SocketChooseHandler extends ByteToMessageDecoder {
    /** 默认暗号长度为23 */
    private static final int MAX_LENGTH = 23;
    /** WebSocket握手的协议前缀 */
    private static final String WEBSOCKET_PREFIX = "GET /";
    @Resource
    private SpringContextUtil springContextUtil;

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        String protocol = getBufStart(in);
        if (protocol.startsWith(WEBSOCKET_PREFIX)) {
            springContextUtil.getBean(PipelineAdd.class).websocketAdd(ctx);

            //对于 webSocket ，不设置超时断开
            ctx.pipeline().remove(IdleStateHandler.class);
//            ctx.pipeline().remove(LengthFieldBasedFrameDecoder.class);
            this.putChannelType(ctx.channel().id(), true);
        }else{
            this.putChannelType(ctx.channel().id(), false);
        }
        in.resetReaderIndex();
        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);
    }

    /**
     *
     * @param channelId
     * @param type
     */
    public void putChannelType(ChannelId channelId,Boolean type){
        if (General.CHANNEL_TYPE_MAP.containsKey(channelId)) {
            log.info("Socket------客户端【" + channelId + "】是否websocket协议："+type);
        } else {
            //保存连接
            General.CHANNEL_TYPE_MAP.put(channelId, type);
            log.info("Socket------客户端【" + channelId + "】是否websocket协议："+type);
        }
    }
}

（5）给***tyServerChannelInitializer初始化类中的***monhandler添加前置处理器

package ***.example.dzx.***ty.qiyanproject.server;

import io.***ty.channel.ChannelHandlerContext;
import io.***ty.handler.codec.http.HttpObjectAggregator;
import io.***ty.handler.codec.http.HttpServerCodec;
import io.***ty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.***ty.handler.stream.ChunkedWriteHandler;
import org.springframework.stereotype.***ponent;

/**
 * @author dzx
 * @ClassName:
 * @Description: 给***tyServerChannelInitializer初始化类中的***monhandler添加前置处理器
 * @date 2023年06月30日 21:31:24
 */
@***ponent
public class PipelineAdd {

    public void websocketAdd(ChannelHandlerContext ctx) {
        System.out.println("PipelineAdd");
        ctx.pipeline().addBefore("***monhandler", "http-codec", new HttpServerCodec());
        ctx.pipeline().addBefore("***monhandler", "aggregator", new HttpObjectAggregator(999999999));
        ctx.pipeline().addBefore("***monhandler", "http-chunked", new ChunkedWriteHandler());
//        ctx.pipeline().addBefore("***monhandler","WebSocketServer***pression",new WebSocketServer***pressionHandler());
        ctx.pipeline().addBefore("***monhandler", "ProtocolHandler", new WebSocketServerProtocolHandler("/ws"));

//        ctx.pipeline().addBefore("***monhandler","StringDecoder",new StringDecoder(CharsetUtil.UTF_8)); // 解码器，将字节转换为字符串
//        ctx.pipeline().addBefore("***monhandler","StringEncoder",new StringEncoder(CharsetUtil.UTF_8));
        // HttpServerCodec：将请求和应答消息解码为HTTP消息
//        ctx.pipeline().addBefore("***monhandler","http-codec",new HttpServerCodec());
//
//        // HttpObjectAggregator：将HTTP消息的多个部分合成一条完整的HTTP消息
//        ctx.pipeline().addBefore("***monhandler","aggregator",new HttpObjectAggregator(999999999));
//
//        // ChunkedWriteHandler：向客户端发送HTML5文件,文件过大会将内存撑爆
//        ctx.pipeline().addBefore("***monhandler","http-chunked",new ChunkedWriteHandler());
//
//        ctx.pipeline().addBefore("***monhandler","WebSocketAggregator",new WebSocketFrameAggregator(999999999));
//
//        //用于处理websocket, /ws为访问websocket时的uri
//        ctx.pipeline().addBefore("***monhandler","ProtocolHandler", new WebSocketServerProtocolHandler("/ws"));

    }
}

（6）编写业务处理器

package ***.example.dzx.***ty.qiyanproject.server;


import ***.example.dzx.***ty.qiyanproject.constants.General;
import io.***ty.buffer.ByteBuf;
import io.***ty.buffer.Unpooled;
import io.***ty.channel.ChannelHandlerContext;
import io.***ty.channel.ChannelId;
import io.***ty.channel.SimpleChannelInboundHandler;
import io.***ty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.***ty.handler.codec.http.websocketx.WebSocketFrame;
import io.***ty.handler.timeout.IdleState;
import io.***ty.handler.timeout.IdleStateEvent;
import io.***ty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

import java.***.I***SocketAddress;
import java.util.Set;
import java.util.stream.Collectors;


/**
 * @author dzx
 * @ClassName:
 * @Description: ***ty服务端处理类
 * @date 2023年06月30日 21:27:16
 */
@Slf4j
public class ***tyServerHandler extends SimpleChannelInboundHandler<Object> {


    //由于继承了SimpleChannelInboundHandler，这个方法必须实现，否则报错
    //但实际应用中，这个方法没被调用
    @Override
    public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buff = (ByteBuf) msg;
        String info = buff.toString(CharsetUtil.UTF_8);
        log.info("收到消息内容：" + info);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // WebSocket消息处理
        String strMsg = "";
        if (msg instanceof WebSocketFrame) {
            log.info("WebSocket消息处理************************************************************");
            strMsg = ((TextWebSocketFrame) msg).text().trim();
            log.info("收到webSocket消息：" + strMsg);
        }
        // Socket消息处理
        else if (msg instanceof ByteBuf) {
            log.info("Socket消息处理=================================");
            ByteBuf buff = (ByteBuf) msg;
            strMsg = buff.toString(CharsetUtil.UTF_8).trim();
            log.info("收到socket消息：" + strMsg);
        }
//        else {
//            strMsg = msg.toString();
//        }
        this.channelWrite(ctx.channel().id(), strMsg);
    }

    /**
     * 有客户端终止连接服务器会触发此函数
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {

        I***SocketAddress insocket = (I***SocketAddress) ctx.channel().remoteAddress();

        String clientIp = insocket.getAddress().getHostAddress();

        ChannelId channelId = ctx.channel().id();

        //包含此客户端才去删除
        if (General.CHANNEL_MAP.containsKey(channelId)) {
            //删除连接
            General.CHANNEL_MAP.remove(channelId);
            System.out.println();
            log.info("Socket------客户端【" + channelId + "】退出***ty服务器[IP:" + clientIp + "--->PORT:" + insocket.getPort() + "]");
            log.info("Socket------连接通道数量: " + General.CHANNEL_MAP.size());
            General.CHANNEL_TYPE_MAP.remove(channelId);
        }
    }


    /**
     * 服务端给客户端发送消息
     */
    public void channelWrite(ChannelId channelId, Object msg) throws Exception {

        ChannelHandlerContext ctx = General.CHANNEL_MAP.get(channelId);

        if (ctx == null) {
            log.info("Socket------通道【" + channelId + "】不存在");
            return;
        }

        if (msg == null || msg == "") {
            log.info("Socket------服务端响应空的消息");
            return;
        }

        //将客户端的信息直接返回写入ctx
        log.info("Socket------服务端端返回报文......【" + channelId + "】" + " :" + (String) msg);
//        ctx.channel().writeAndFlush(msg);
//        ctx.writeAndFlush(msg);
        //刷新缓存区
//        ctx.flush();
        //过滤掉当前通道id
        Set<ChannelId> channelIdSet = General.CHANNEL_MAP.keySet().stream().filter(id -> !id.asLongText().equalsIgnoreCase(channelId.asLongText())).collect(Collectors.toSet());
        //广播消息到客户端
        for (ChannelId id : channelIdSet) {
            //是websocket协议
            Boolean aBoolean = General.CHANNEL_TYPE_MAP.get(id);
            if(aBoolean!=null && aBoolean){
                General.CHANNEL_MAP.get(id).channel().writeAndFlush(new TextWebSocketFrame((String) msg));
            }else {
                ByteBuf byteBuf = Unpooled.copiedBuffer(((String) msg).getBytes());
                General.CHANNEL_MAP.get(id).channel().writeAndFlush(byteBuf);
            }
        }
    }

    /**
     * 处理空闲状态事件
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        String socketString = ctx.channel().remoteAddress().toString();

        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                log.info("Socket------Client: " + socketString + " READER_IDLE 读超时");
                ctx.disconnect();
            } else if (event.state() == IdleState.WRITER_IDLE) {
                log.info("Socket------Client: " + socketString + " WRITER_IDLE 写超时");
                ctx.disconnect();
            } else if (event.state() == IdleState.ALL_IDLE) {
                log.info("Socket------Client: " + socketString + " ALL_IDLE 总超时");
                ctx.disconnect();
            }
        }
    }

    /**
     * @DESCRIPTION: 发生异常会触发此函数
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
        log.error("Socket------" + ctx.channel().id() + " 发生了错误,此连接被关闭" + "此时连通数量: " + General.CHANNEL_MAP.size(),cause);
    }
}

 （7）spring上下文工具类

package ***.example.dzx.***ty.qiyanproject.***ty1;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.***ponent;

/**
 * @author dzx
 * @ClassName:
 * @Description: spring容器上下文工具类
 * @date 2023年06月30日 21:30:02
 */
@***ponent
public class SpringContextUtil implements ApplicationContextAware {

    private static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringContextUtil.applicationContext = applicationContext;
    }

    /**
     * @Description: 获取spring容器中的bean, 通过bean类型获取
     */
    public static <T> T getBean(Class<T> beanClass) {
        return applicationContext.getBean(beanClass);
    }

}

（8）编写全局map常量类

package ***.example.dzx.***ty.qiyanproject.constants;

import io.***ty.channel.ChannelHandlerContext;
import io.***ty.channel.ChannelId;

import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 500007
 * @ClassName:
 * @Description:
 * @date 2023年07月02日 19:12:42
 */
public class General {

    /**
     * 管理一个全局map，保存连接进服务端的通道数量
     */
    public static final ConcurrentHashMap<ChannelId, ChannelHandlerContext> CHANNEL_MAP = new ConcurrentHashMap<>();

    /**
     * 管理一个全局mao, 报存连接进服务器的各个通道类型
     */
    public static final ConcurrentHashMap<ChannelId, Boolean> CHANNEL_TYPE_MAP = new ConcurrentHashMap<>();

}

三、***ty客户端

（1）编写***ty客户端，用于测试向服务端的消息发送

package ***.example.dzx.***ty.qiyanproject.Socket;

import io.***ty.bootstrap.Bootstrap;
import io.***ty.channel.*;
import io.***ty.channel.nio.NioEventLoopGroup;
import io.***ty.channel.socket.SocketChannel;
import io.***ty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author dzx
 * @ClassName:
 * @Description: ***ty客户端
 * @date 2023年06月30日 21:30:02
 */
public class SocketClient {
    // 服务端IP
    static final String HOST = System.getProperty("host", "127.0.0.1");

    // 服务端开放端口
    static final int PORT = Integer.parseInt(System.getProperty("port", "7777"));

    // 日志打印
    private static final Logger LOGGER = LoggerFactory.getLogger(SocketClient.class);

    // 主函数启动
    public static void main(String[] args) throws InterruptedException {
        sendMessage("我是客户端，我发送了一条数据给***ty服务端。。");
    }

    /**
     * 核心方法（处理：服务端向客户端发送的数据、客户端向服务端发送的数据）
     */
    public static void sendMessage(String content) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new SocketChannelInitializer());
                        }
                    });

            ChannelFuture future = b.connect(HOST, PORT).sync();
            for (int i = 0; i < 3; i++) {
                future.channel().writeAndFlush(content);
                Thread.sleep(2000);
            }
            // 程序阻塞
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

}

（2）编写***ty客户端初始化处理器

package ***.example.dzx.***ty.qiyanproject.Socket;


import io.***ty.channel.ChannelInitializer;
import io.***ty.channel.ChannelPipeline;
import io.***ty.channel.socket.SocketChannel;
import io.***ty.handler.codec.string.StringDecoder;
import io.***ty.handler.codec.string.StringEncoder;
import io.***ty.util.CharsetUtil;


/**
 * @author dzx
 * @ClassName:
 * @Description: ***ty客户端初始化时设置出站和入站的编码器和解码器
 * @date 2023年06月30日 21:30:02
 */
public class SocketChannelInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel channel) throws Exception {
        ChannelPipeline p = channel.pipeline();
        p.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
        p.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
        p.addLast(new SocketHandler());
    }
}

（3）***ty客户端业务处理器，用于接收并处理服务端发送的消息数据

package ***.example.dzx.***ty.qiyanproject.Socket;


import io.***ty.channel.ChannelHandlerContext;
import io.***ty.channel.ChannelInboundHandlerAdapter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.***.I***SocketAddress;


/**
 * @author dzx
 * @ClassName:
 * @Description: ***ty客户端处理器
 * @date 2023年06月30日 21:30:02
 */
@Slf4j
public class SocketHandler extends ChannelInboundHandlerAdapter {

    // 日志打印
    private static final Logger LOGGER = LoggerFactory.getLogger(SocketHandler.class);

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        LOGGER.debug("SocketHandler Active（客户端）");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        LOGGER.debug("####接收服务端发送过来的消息####");
        LOGGER.debug("SocketHandler read Message：" + msg);
        //获取服务端连接的远程地址
        I***SocketAddress insocket = (I***SocketAddress) ctx.channel().remoteAddress();
        //获取服务端的IP地址
        String clientIp = insocket.getAddress().getHostAddress();
        //获取服务端的端口号
        int clientPort = insocket.getPort();
        log.info("***ty服务端[IP:" + clientIp + "--->PORT:" + clientPort + "]");

    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        LOGGER.debug("####客户端断开连接####");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

至此，***ty服务端和***ty客户端都编写完毕，我们可以来进行测试了。

四、测试

（1）前端 websocket 向 后端***Assist测试工具发送消息

 在前端窗口向后端 发送了一个 22222的 字符串，后端测试工具成功接收到消息并展示在对话框中。

（2）后端***Assist向 前端 websocket 发送消息

  在后端窗口向前端 发送了一个{"deviceId":"11111","deviceName":"qz-01","deviceStatus":"2"}的 字符串，前端测试工具成功接收到消息并展示在对话框中。

五、代码仓库地址

完整项目已上传至gitee仓库，请点击下方传送门自行获取，一键三连！！

https://gitee.***/dzxmy/***ty-web-socketd-dnamic

无法访问就点击下方传送门去我的资源下载即可

https://download.csdn.***/download/qq_31905135/88044942?spm=1001.2014.3001.5503