package com.tongran.agent.client.netty; import com.tongran.agent.client.netty.config.BaseNettyConfig; import com.tongran.agent.client.utils.AssertLog; import io.netty.bootstrap.AbstractBootstrap; import io.netty.bootstrap.Bootstrap; 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.NioChannelOption; import io.netty.channel.socket.nio.NioDatagramChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.util.ResourceLeakDetector; import io.netty.util.concurrent.DefaultEventExecutorGroup; import io.netty.util.concurrent.DefaultThreadFactory; import io.netty.util.concurrent.EventExecutorGroup; import io.netty.util.concurrent.Future; /** * 基础Netty服务 */ public abstract class BaseNettyServer { protected boolean isRunning; protected BaseNettyConfig config; protected EventLoopGroup bossGroup; protected EventLoopGroup workerGroup; protected EventExecutorGroup businessGroup; protected BaseNettyServer(BaseNettyConfig config) { this.config = config; } protected AbstractBootstrap initializeTcp() { bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY)); workerGroup = new NioEventLoopGroup(config.workerCore, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY)); if (config.businessCore > 0) { businessGroup = new DefaultEventExecutorGroup(config.businessCore); } ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childOption(ChannelOption.SO_REUSEADDR, true) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(NioChannelOption.TCP_NODELAY, true) .childHandler(config.hander); //内存泄漏检测 开发推荐PARANOID 线上SIMPLE ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.SIMPLE); return serverBootstrap; } protected AbstractBootstrap initializeUdp() { bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY)); if (config.businessCore > 0) { businessGroup = new DefaultEventExecutorGroup(config.businessCore); } return new Bootstrap() .group(bossGroup).channel(NioDatagramChannel.class) .option(NioChannelOption.SO_REUSEADDR, true) .option(NioChannelOption.SO_RCVBUF, 1024 * 1024 * 50) .handler(config.hander); } public synchronized boolean start() { if (!config.enable) { return false; } if (isRunning) { AssertLog.info("======{}已经启动,port:{}======", config.name, config.port); return isRunning; } AbstractBootstrap bootstrap = config.isTcp ? initializeTcp() : initializeUdp(); ChannelFuture future = bootstrap.bind(config.port).awaitUninterruptibly(); future.channel().closeFuture().addListener(f -> { if (isRunning) { stop(); } }); if (future.cause() != null) { AssertLog.error("===启动失败===", future.cause()); } if (isRunning = future.isSuccess()) { AssertLog.info("\n\n\t\t\t\t\t\t\t\t======{}启动成功,port:{}======\n", config.name, config.port); } return isRunning; } public synchronized void stop() { if (!config.enable) { return; } isRunning = false; try { Future future = this.workerGroup.shutdownGracefully().await(); if (!future.isSuccess()) { AssertLog.error("workerGroup 无法正常停止:{}", future.cause()); } future = this.bossGroup.shutdownGracefully().await(); if (!future.isSuccess()) { AssertLog.error("bossGroup 无法正常停止:{}", future.cause()); } future = this.businessGroup.shutdownGracefully().await(); if (!future.isSuccess()) { AssertLog.error("businessGroup 无法正常停止:{}", future.cause()); } } catch (InterruptedException e) { e.printStackTrace(); } AssertLog.info("\n\n\t\t\t\t\t\t\t\t======{} 已经停止,port:{}======\n", config.name, config.port); } }