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

这篇具有很好参考价值的文章主要介绍了SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦)。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

 

前言

技术栈

功能展示

一、springboot项目添加netty依赖

二、netty服务端

三、netty客户端

四、测试

五、代码仓库地址


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

前言

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

技术栈

        后端

  • springboot 
  • netty

        前端

  • 前端websocket

功能展示

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

SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦),netty,springboot,websocket,spring boot,websocket,netty

 后端我们可以使用网络测试工具NetAssist 进行响应测试SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦),netty,springboot,websocket,spring boot,websocket,netty

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

一、springboot项目添加netty依赖

<?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>com.example.dzx.netty</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.netty</groupId>
            <artifactId>netty-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>

二、netty服务端

 (1)netty服务启动类

package com.example.dzx.netty.qiyanproject.server;

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.stereotype.Component;

import java.net.InetSocketAddress;

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

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

        /*
         * 在Netty中,事件循环组是一组线程池,用于处理网络事件,例如接收客户端连接、读写数据等操作。
         * 它由两个部分组成: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 NettyServerChannelInitializer())
//                    .childHandler(new com.ccp.dev.system.netty.NettyServerChannelInitializer())
                    .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.isSuccess()) {
                    System.out.println("Netty服务启动成功");
                } else {
                    System.out.println("Netty服务启动失败");
                }
            });
            log.info("Netty服务开始监听端口: " + address.getPort());
            //关闭channel和块,直到它被关闭
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("启动Netty服务器时出错", e);
        } finally {
            //释放资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

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

package com.example.dzx.netty.qiyanproject.server;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.stereotype.Component;


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

//    private FullHttpResponse createCorsResponseHeaders() {
//        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
//
//        // 设置允许跨域访问的响应头
//        response.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_ORIGIN, "*");
//        response.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_METHODS, "GET, POST, PUT, DELETE");
//        response.headers().set(HttpHeaderNames.ACCESS_CONTROL_ALLOW_HEADERS, "Content-Type, Authorization");
//        response.headers().set(HttpHeaderNames.ACCESS_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("commonhandler",new NettyServerHandler());
    }
}

(3) 编写新建连接处理器

package com.example.dzx.netty.qiyanproject.server;

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

import java.net.InetSocketAddress;

/**
 * @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) {
        //获取客户端连接的远程地址
        InetSocketAddress insocket = (InetSocketAddress) 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 + "】连接netty服务器[IP:" + clientIp + "--->PORT:" + clientPort + "]");
            log.info("Socket------连接通道数量: " + General.CHANNEL_MAP.size());
        }
    }
}

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

package com.example.dzx.netty.qiyanproject.server;

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

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

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

/**
 * @author dzx
 * @ClassName:
 * @Description:  协议初始化解码器.用来判定实际使用什么协议,以用来处理前端websocket或者后端netty客户端的连接或通信
 * @date 2023年06月30日 21:31:24
 */
@Component
@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)给NettyServerChannelInitializer初始化类中的commonhandler添加前置处理器

package com.example.dzx.netty.qiyanproject.server;

import io.netty.channel.ChannelHandlerContext;
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 org.springframework.stereotype.Component;

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

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

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

    }
}

(6)编写业务处理器

package com.example.dzx.netty.qiyanproject.server;


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

import java.net.InetSocketAddress;
import java.util.Set;
import java.util.stream.Collectors;


/**
 * @author dzx
 * @ClassName:
 * @Description: netty服务端处理类
 * @date 2023年06月30日 21:27:16
 */
@Slf4j
public class NettyServerHandler 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) {

        InetSocketAddress insocket = (InetSocketAddress) 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 + "】退出netty服务器[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 com.example.dzx.netty.qiyanproject.netty1;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

/**
 * @author dzx
 * @ClassName:
 * @Description: spring容器上下文工具类
 * @date 2023年06月30日 21:30:02
 */
@Component
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 com.example.dzx.netty.qiyanproject.constants;

import io.netty.channel.ChannelHandlerContext;
import io.netty.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<>();

}

三、netty客户端

(1)编写netty客户端,用于测试向服务端的消息发送

package com.example.dzx.netty.qiyanproject.Socket;

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

/**
 * @author dzx
 * @ClassName:
 * @Description: netty客户端
 * @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("我是客户端,我发送了一条数据给netty服务端。。");
    }

    /**
     * 核心方法(处理:服务端向客户端发送的数据、客户端向服务端发送的数据)
     */
    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)编写netty客户端初始化处理器

package com.example.dzx.netty.qiyanproject.Socket;


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


/**
 * @author dzx
 * @ClassName:
 * @Description: netty客户端初始化时设置出站和入站的编码器和解码器
 * @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)netty客户端业务处理器,用于接收并处理服务端发送的消息数据

package com.example.dzx.netty.qiyanproject.Socket;


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

import java.net.InetSocketAddress;


/**
 * @author dzx
 * @ClassName:
 * @Description: netty客户端处理器
 * @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);
        //获取服务端连接的远程地址
        InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
        //获取服务端的IP地址
        String clientIp = insocket.getAddress().getHostAddress();
        //获取服务端的端口号
        int clientPort = insocket.getPort();
        log.info("netty服务端[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();
    }
}

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

四、测试

(1)前端 websocket 向 后端NetAssist测试工具发送消息

SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦),netty,springboot,websocket,spring boot,websocket,netty

 SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦),netty,springboot,websocket,spring boot,websocket,netty

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

(2)后端NetAssist向 前端 websocket 发送消息

SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦),netty,springboot,websocket,spring boot,websocket,netty

 SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦),netty,springboot,websocket,spring boot,websocket,netty

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

五、代码仓库地址

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

https://gitee.com/dzxmy/netty-web-socketd-dnamic

SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦),netty,springboot,websocket,spring boot,websocket,netty

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

https://download.csdn.net/download/qq_31905135/88044942?spm=1001.2014.3001.5503文章来源地址https://www.toymoban.com/news/detail-532767.html

到了这里,关于SpringBoot项目整合WebSocket+netty实现前后端双向通信(同时支持前端webSocket和socket协议哦)的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • SpringBoot+Netty+Websocket实现消息推送

    这样一个需求:把设备异常的状态每10秒推送到页面并且以弹窗弹出来,这个时候用Websocket最为合适,今天主要是后端代码展示。 添加依赖 定义netty端口号 netty服务器 Netty配置 管理全局Channel以及用户对应的channel(推送消息) 管道配置 自定义CustomChannelHandler 推送消息接口及

    2024年02月04日
    浏览(49)
  • Netty系列(一):Springboot整合Netty,自定义协议实现

    Netty是由JBOSS提供的一个java开源框架,现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序。 也就是说,Netty 是一个基于NIO的客户、服务器端的编程框架,使用Netty 可以确保你快速和简单

    2023年04月25日
    浏览(55)
  • Springboot中使用netty 实现 WebSocket 服务

    依赖 创建启动类 创建WebSocket 服务 WsServerInitialzer 初始化 创建信息ChatHandler 处理类

    2024年02月14日
    浏览(38)
  • Springboot整合Netty,自定义协议实现

    新建springboot项目,并在项目以来中导入netty包,用fastjson包处理jsonStr。 创建netty相关配置信息文件 yml配置文件—— application.yml netty配置实体类—— NettyProperties 与yml配置文件绑定 通过 @ConfigurationProperties(prefix = \\\"netty\\\") 注解读取配置文件中的netty配置,通过反射注入值,需要在

    2024年02月06日
    浏览(39)
  • Springboot整合Netty实现RPC服务器

    try { ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(boss, worker) .childHandler(new ChannelInitializer() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast(new IdleStateHandler(0, 0, 60)); pipeline.addLast(new JsonDecoder()); pipeline.addLast(new JsonEnco

    2024年04月09日
    浏览(43)
  • 基于Springboot用Netty实现WebSocket及用户身份校验

    说在前头,文本主要参考: SpringBoot+WebSocket+Netty实现消息推送 Netty-11-channelHandler的生命周期 springboot整合netty指北 首先 需要了解下channel建立的生命周期 ChannelHandler的顺序如下: 注意本次实现的重点是:在建立websocket时从请求标头header或者第一次消息对话时获取用户信息(如jw

    2024年02月04日
    浏览(42)
  • SpringBoot+Netty+Vue+Websocket实现在线推送/聊天系统

    ok,那么今天的话也是带来这个非常常用的一个技术,那就是咱们完成nutty的一个应用,今天的话,我会介绍地很详细,这样的话,拿到这个博文的代码就基本上可以按照自己的想法去构建自己的一个在线应用了。比如聊天,在线消息推送之类的。其实一开始我原来的想法做在

    2024年02月03日
    浏览(42)
  • 基于Springboot+WebSocket+Netty实现在线聊天、群聊系统

    此文主要实现在好友添加、建群、聊天对话、群聊功能,使用Java作为后端语言进行支持,界面友好,开发简单。 2.1、下载安装IntelliJ IDEA(后端语言开发工具),Mysql数据库,微信Web开发者工具。 1.创建maven project 先创建一个名为SpringBootDemo的项目,选择【New Project】 然后在弹出

    2024年02月14日
    浏览(42)
  • netty学习(3):SpringBoot整合netty实现多个客户端与服务器通信

    创建一个SpringBoot工程,然后创建三个子模块 整体工程目录:一个server服务(netty服务器),两个client服务(netty客户端) pom文件引入netty依赖,springboot依赖 NettySpringBootApplication NettyServiceHandler SocketInitializer NettyServer NettyStartListener application.yml Client1 NettyClientHandler SocketInitializ

    2024年02月11日
    浏览(59)
  • Spring Boot整合WebSocket实现实时通信,前端实时通信,前后端实时通信

    实时通信在现代Web应用中扮演着越来越重要的角色,无论是在线聊天、股票价格更新还是实时通知,WebSocket都是实现这些功能的关键技术之一。Spring Boot作为一个简化企业级应用开发的框架,其对WebSocket的支持也非常友好。本文将详细介绍如何在Spring Boot中整合WebSocket,实现一

    2024年04月27日
    浏览(39)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包