SpringBoot集成Netty实现Socket通信

SpringBoot集成Netty实现Socket通信（Spring容器托管）

一、开始

注：Socket与WebSocket并不一样。其中Sokcet通过ip:port进行通信，一个ip+port只能实现一条Socket连接，；WebSocket通过ip:port/路径，进行通信，在ip与port一致的情况下，不同的路径可以实现不同的Socket连接

SpringBoot父版本

<parent>
    <artifactId>spring-boot-starter-parent</artifactId>
    <groupId>org.springframework.boot</groupId>
    <version>2.5.12</version>
    <relativePath/>
</parent>

引入dependencies（需要Web服务的话将spring-boot-starter改成spring-boot-starter-web）

• netty需要独占端口，与web容器端口不同

<!-- spring -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter</artifactId>
</dependency>

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

二、代码配置

application.yml配置netty端口

# netty
netty:
port: 9988

NettyServer.java（Spring容器托管Netty）

@Slf4j
@Component
public class NettyServer implements ApplicationRunner, ApplicationListener<ContextClosedEvent>, ApplicationContextAware {
    // 端口
    @Value("${netty.port}")
    private Integer port;

    // spring容器上下文
    private ApplicationContext applicationContext;

    // netty连接channel
    private Channel serverChannel;

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

    @Override
    public void run(ApplicationArguments args) throws Exception {
        // 连接事件线程组
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        // 读写事件线程组
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            //设置两个线程组boosGroup和workerGroup
            bootstrap.group(bossGroup, workerGroup)
                //设置服务端通道实现类型
                .channel(NioServerSocketChannel.class)
                // 初始化通道对象
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        log.info("------netty新连接：{}------", ch.remoteAddress());
                        // MyHandlerAdapter.class是自定义的数据解释执行器
                        ch.pipeline()
                            .addLast(applicationContext.getBean(MyHandlerAdapter.class));
                    }
                })
                //设置线程队列可等待的连接个数
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childOption(ChannelOption.SO_KEEPALIVE, true);

            Channel channel = bootstrap.bind(port).sync().channel();
            this.serverChannel = channel;
            log.info("------netty服务端启动，端口：{}------", port);
            // 关闭服务器通道
            channel.closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully(); // 释放线程池资源
            bossGroup.shutdownGracefully();
        }
    }

    // 监听容器关闭事件
    @Override
    public void onApplicationEvent(ContextClosedEvent event) {
        if (this.serverChannel != null) {
            this.serverChannel.close();
        }
        log.info("----------Socket服务停止----------");
    }
}

MyHandlerAdapter.java（Netty消息通信核心部分，自定义实现）

@Slf4j
@Component
// 防止netty在spring托管下多连接抛异常
@ChannelHandler.Sharable
    public class MyHandlerAdapter extends ChannelInboundHandlerAdapter {
        
        @Resource
        private MsgHeaderDecoder decoder;

        @Resource
        private MsgTypeAndOpt msgTypeAndOpt;
        
        
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            String id = ctx.channel().id().asLongText();
            log.info("连接，id:" + id);
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            log.info("断开");
            ctx.channel().close();
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            // 接收到服务端发送的ByteBuf，进行解析
            ByteBuf buf = (ByteBuf) msg;
            byte[] bytes = new byte[buf.readableBytes()];
            buf.readBytes(bytes);
            // 根据自定义操作编写接收到数据后的实现部分
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            log.info("用户事件");
        }


        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
            log.info("发生异常");
            // 全局抛出异常
            e.printStackTrace();
            ctx.channel().close();
        }
    }