【实现netty服务同时监听多个端口，处理多套协议】

• Netty
• • 1. 实现Netty服务端
  • 2. 通道初始化
  • 3. 创建对应的解析器和编码器
  • 4. 编写服务端处理器
  • 5. Netty服务启动类
  • 结尾

netty一个提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。
在上一篇文章中介绍了Netty怎么去实现和硬件双工通信：https://blog.csdn.net/GBK_8/article/details/123256716?spm=1001.2014.3001.5502

@Slf4j
@Component
@RefreshScope
public class NettyServerOfLims {
    
    @Autowired
    private PortDefinition portDefinition;

    ChannelFuture future = null;

    NioEventLoopGroup boss = null;

    NioEventLoopGroup worker = null;

    ServerBootstrap bootstrap = new ServerBootstrap();

    public void start() {
        boss = new NioEventLoopGroup();
        worker = new NioEventLoopGroup();
        //1.绑定group
        bootstrap.group(boss, worker)
                //2.设置并绑定Reactor线程池：EventLoopGroup，EventLoop就是处理所有注册到本线程的Selector上面的Channel
                .channel(NioServerSocketChannel.class)
                //3.保存连接数
                .option(ChannelOption.SO_BACKLOG, 1024)
                //4.有数据立即发送
                //.option(ChannelOption.SO_SNDBUF, 4 * 1024)
                //.option(ChannelOption.SO_RCVBUF, 4 * 1024)
                .handler(new LoggingHandler(LogLevel.INFO))
                //5.保持连接
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                //6.handler作用于boss -- childHandler作用于worker
                .childHandler(new SocketChannelInitHandler(portDefinition.getPort()));

        Map ports = portDefinition.getPort();
        log.info("netty服务器在{}端口启动监听", JSONObject.toJSONString(ports));
        try {
             
            for (Map.Entry p : ports.entrySet()) {
                final int port = p.getKey();
                // 绑定端口
                ChannelFuture future1 = bootstrap.bind(new InetSocketAddress(port)).sync();
                future1.addListener(future -> {
                    if (future.isSuccess()) {
                        log.info("netty 启动成功，端口：{}", port);
                    } else {
                        log.info("netty 启动失败，端口：{}", port);
                    }
                });
                future1.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> future1.channel().close());
            }
        } catch (Exception e) {
            log.error("netty 启动时发生异常-------{}", e);
        }
    }

    @PreDestroy
    public void stop() {
        if (future != null) {
            future.channel().close().addListener(ChannelFutureListener.CLOSE);
            future.awaitUninterruptibly();
            boss.shutdownGracefully();
            worker.shutdownGracefully();
            future = null;
            log.info(" 服务关闭 ");
        }
    }
}

端口配置类

@Data
@Configuration
@ConfigurationProperties(prefix = "netty")
public class PortDefinition {

    Map port;
}

@Slf4j
public class SocketChannelInitHandler extends ChannelInitializer {


    
    public static final Map CHANNEL_MAP = new ConcurrentHashMap<>();


    Map ports;

    public SocketChannelInitHandler(Map ports) {
        this.ports = ports;
    }

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //每次连接上来 对通道进行保存
        CHANNEL_MAP.put(socketChannel.id(), socketChannel.pipeline());
        ChannelPipeline pipeline = socketChannel.pipeline();
        int port = socketChannel.localAddress().getPort();
        String type = ports.get(port);
        //不同类型连接，处理链中加入不同处理协议
        switch (type) {
            case GatewayType.LIMS:
                pipeline.addLast(
                        new AMsgDecoder(),
                        new AMsgEncoder(),
                        new AServerHandler());
                break;
            case GatewayType.EXPLOSION_PROOF:
                //防爆
                pipeline.addLast(
                        new BMsgDecoder(),
                        new BMsgEncoder(),
                        new BServerHandler());
                break;
            case GatewayType.NON_EXPLOSION_PROOF:
                pipeline.addLast(
                        new CExpMsgDecoder(),
                        new CExpMsgEncoder(),
                        new CExpServerHandler());
                break;
            default:
                log.error("当前网关类型并不存在于配置文件中，无法初始化通道");
                break;
        }
        pipeline.addLast(
                new StringEncoder(StandardCharsets.UTF_8),
                new StringDecoder(StandardCharsets.UTF_8));
    }
}

3.1 信息解析器

@Slf4j
public class AMsgDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List