SpringBoot 整合 Netty 多端口監(jiān)聽的操作方法

更新時(shí)間：2023年10月17日 10:51:28 作者：帥氣Dee海綿寶寶

Netty提供異步的、基于事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用程序框架,用以快速開發(fā)高性能、高可靠性的網(wǎng)絡(luò) IO 程序,是目前最流行的 NIO 框架,這篇文章主要介紹了SpringBoot 整和 Netty 并監(jiān)聽多端口,需要的朋友可以參考下

SpringBoot 整合 Netty 并監(jiān)聽多端口

Netty 是由 JBOSS 提供的一個(gè) Java 開源框架。Netty 提供異步的、基于事件驅(qū)動(dòng)的網(wǎng)絡(luò)應(yīng)用程序框架，用以快速開發(fā)高性能、高可靠性的網(wǎng)絡(luò) IO 程序,是目前最流行的 NIO 框架，Netty 在互聯(lián)網(wǎng)領(lǐng)域、大數(shù)據(jù)分布式計(jì)算領(lǐng)域、游戲行業(yè)、通信行業(yè)等獲得了廣泛的應(yīng)用，知名的 Elasticsearch 、Dubbo 框架內(nèi)部都采用了 Netty。

1.依賴

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

2.PortDefinition

讀取yml 配置中的多端口配置

//netty:
//  port: {8300: A, 8500: B}
@Data
@Configuration
@ConfigurationProperties(prefix = "netty")
public class PortDefinition {

    Map<Integer, String> port;
}

3.GatewayType

多端口判斷使用的常量類

public class GatewayType {
    //個(gè)人設(shè)備
    public final static String GERNESHEBEI_SHOUHUA="A";
}

4.NettyServer

實(shí)現(xiàn)Netty服務(wù)端

@Slf4j
@RefreshScope
@Component
public class NettyServer {
    @Autowired
    private PortDefinition portDefinition;
    public void start() throws InterruptedException {
        /**
         * 創(chuàng)建兩個(gè)線程組 bossGroup 和 workerGroup
         * bossGroup 只是處理連接請(qǐng)求，真正的和客戶端業(yè)務(wù)處理，會(huì)交給 workerGroup 完成
         *  兩個(gè)都是無線循環(huán)
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //創(chuàng)建服務(wù)器端的啟動(dòng)對(duì)象，配置參數(shù)
            ServerBootstrap bootstrap = new ServerBootstrap();
            //設(shè)置兩個(gè)線程組
            bootstrap.group(bossGroup, workerGroup)
                    //使用NioServerSocketChannel 作為服務(wù)器的通道實(shí)現(xiàn)
                    .channel(NioServerSocketChannel.class)
                    //設(shè)置線程隊(duì)列得到連接個(gè)數(shù)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    //設(shè)置保持活動(dòng)連接狀態(tài)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //通過NoDelay禁用Nagle,使消息立即發(fā)出去，不用等待到一定的數(shù)據(jù)量才發(fā)出去
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    //可以給 bossGroup 加個(gè)日志處理器
                    .handler(new LoggingHandler(LogLevel.INFO))
					//監(jiān)聽多個(gè)端口
                    .childHandler(new SocketChannelInitHandler(portDefinition.getPort()))
            ;
//            監(jiān)聽多個(gè)端口
            Map<Integer, String> ports = portDefinition.getPort();
            log.info("netty服務(wù)器在{}端口啟動(dòng)監(jiān)聽", JSONObject.toJSONString(ports));
            for (Map.Entry<Integer, String> p : ports.entrySet()) {
                final int port = p.getKey();
                // 綁定端口
                ChannelFuture cf = bootstrap.bind(new InetSocketAddress(port)).sync();
                if (cf.isSuccess()) {
                    log.info("netty 啟動(dòng)成功，端口：{}", port);
                } else {
                    log.info("netty 啟動(dòng)失敗，端口：{}", port);
                }
                //對(duì)關(guān)閉通道進(jìn)行監(jiān)聽
                cf.channel().closeFuture().sync();
            }
        } finally {
            //發(fā)送異常關(guān)閉
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

5.SocketChannelInitHandler

根據(jù)多端口去判斷去執(zhí)行那個(gè)業(yè)務(wù)的 Handler 方法

@Slf4j
public class SocketChannelInitHandler extends ChannelInitializer<SocketChannel> {
    /**
     * 用來存儲(chǔ)每個(gè)連接上來的設(shè)備
     */
    public static final Map<ChannelId, ChannelPipeline> CHANNEL_MAP = new ConcurrentHashMap<>();
    /**
     * 端口信息，用來區(qū)分這個(gè)端口屬于哪種類型的連接 如：8300 屬于 A
     */
    Map<Integer, String> ports;
    public SocketChannelInitHandler(Map<Integer, String> ports) {
        this.ports = ports;
    }
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //每次連接上來 對(duì)通道進(jìn)行保存
        CHANNEL_MAP.put(socketChannel.id(), socketChannel.pipeline());
        ChannelPipeline pipeline = socketChannel.pipeline();
        int port = socketChannel.localAddress().getPort();
        String type = ports.get(port);
        pipeline.addLast(new StringEncoder(StandardCharsets.UTF_8));
        pipeline.addLast(new StringDecoder(StandardCharsets.UTF_8));
        pipeline.addLast(new LengthFieldBasedFrameDecoder(24*1024,0,2));
        log.info("【initChannel】端口號(hào)： "+port+"  類型： "+type);
        //不同類型連接，處理鏈中加入不同處理協(xié)議
        switch (type) {
            case GatewayType.GERNESHEBEI_SHOUHUA:
                //手環(huán)
                pipeline.addLast(new NettyServerHandler());
                break;
            default:
                log.error("當(dāng)前網(wǎng)關(guān)類型并不存在于配置文件中，無法初始化通道");
                break;
        }
    }
}

6.NettyServerHandler

業(yè)務(wù)員的Handler 方法

@Slf4j
public class NettyServerHandler extends SimpleChannelInboundHandler<Object> {
    //private static final Logger log = LoggerFactory.getLogger(NettyServerHandler.class);
    protected void channelRead0(ChannelHandlerContext context, Object obj) throws Exception {
        log.info(">>>>>>>>>>>服務(wù)端接收到客戶端的消息：{}",obj);
        String message = (String)obj;
		//之后寫自己的業(yè)務(wù)邏輯即可
        String b=message.replace("[", "")
                .replace("]","");
        String[] split1 = b.split("\\*");
        String key = split1[1];
        SocketChannel socketChannel = (SocketChannel) context.channel();
		//調(diào)用業(yè)務(wù)代碼類并執(zhí)行
        WristWatchSocket.runSocket(key,message,socketChannel,true);
        ReferenceCountUtil.release(obj);
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

7.WristWatchSocket 業(yè)務(wù)代碼調(diào)用,根據(jù)自己的業(yè)務(wù)需要

@Slf4j
//@Data
public class WristWatchSocket{
    //業(yè)務(wù)代碼
    public static void runSocket(String key, String message, SocketChannel socketChannel, Boolean isNotFirst){
        try {
            if(message==null||message.trim().equals("")){
                return;
            }
            String restr="測(cè)試發(fā)送";
            if(!"".equals(restr)){
                socketChannel.writeAndFlush(restr);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

8.配置文件配置多端口

netty:
  port: {8300: A, 8500: B}

9.配置Netty 啟動(dòng)

在啟動(dòng)類中配置

@EnableAsync
@EnableSwagger2
@EnableFeignClients
@EnableTransactionManagement
@Slf4j
@MapperScan({"com.yuandian.platform.mapper"})
@SpringBootApplication(exclude={DataSourceAutoConfiguration.class})
public class YunYiplatformApplication {
    public static void main(String[] args) {
        ApplicationContext run = SpringApplication.run(YunYiplatformApplication.class, args);
        log.info("\n\n【【【【平臺(tái)成功啟動(dòng)！】】】】\n");
        //netty 啟動(dòng)配置
        try {
            run.getBean(NettyServer.class).start();
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

截圖