Merge branch 'for-yx' of http://120.77.240.215:9701/tianlai3/ioc_sixiang_license into for-yx

license/src/main/java/iot/sixiang/license/LicenseApplication.java

@@ -4,11 +4,12 @@ import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.web.servlet.ServletComponentScan;
+import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.web.servlet.config.annotation.ResourceHandlerRegistry;
 import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
 
-
+@EnableAsync
 @ServletComponentScan(basePackages ="iot.sixiang.license")
 @SpringBootApplication
 @EnableScheduling

license/src/main/java/iot/sixiang/license/config/ThreadPoolConfig.java

+package iot.sixiang.license.config;
+
+/**
+ * Title: ThreadPoolConfig
+ * Description: TODO
+ *
+ * @author tianlai3
+ * @date 2022-07-16 20:05:38
+ */
+import lombok.Data;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ThreadPoolExecutor;
+@Data
+@Configuration
+@EnableAsync
+public class ThreadPoolConfig {
+
+
+    private static final int corePoolSize = 1;   // 核心线程数（默认线程数）
+    private static final int maxPoolSize = 2;   // 最大线程数
+    private static final int keepAliveTime = 10;  // 允许线程空闲时间（单位：默认为秒）
+    private static final int queueCapacity = 2; // 缓冲队列数
+    /**
+     * 默认异步线程池
+     * @return
+     */
+    @Bean("taskExecutor")
+    public Executor taskExecutor(){
+        ThreadPoolTaskExecutor pool = new ThreadPoolTaskExecutor();
+        pool.setThreadNamePrefix("threadPoll-");
+        pool.setCorePoolSize(corePoolSize);
+        pool.setMaxPoolSize(maxPoolSize);
+        pool.setKeepAliveSeconds(keepAliveTime);
+        pool.setQueueCapacity(queueCapacity);
+        pool.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        // 初始化
+        pool.initialize();
+        return pool;
+    }
+}

license/src/main/java/iot/sixiang/license/device/AsyncTcpServer.java

+package iot.sixiang.license.device;
+
+import iot.sixiang.license.net.TcpServer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Component;
+
+/**
+ * Title: AsyncTcpServer
+ * Description: TODO
+ *
+ * @author tianlai3
+ * @date 2022-07-16 20:30:51
+ */
+@Component
+@Slf4j
+public class AsyncTcpServer {
+
+    @Async("taskExecutor")
+    public void start(int port, DeviceChannelInitializer channelInitializer) {
+        TcpServer server = new TcpServer(port, channelInitializer);
+        server.start();
+    }
+}

license/src/main/java/iot/sixiang/license/device/DeviceManager.java

@@ -5,7 +5,6 @@ import iot.sixiang.license.model.PageInfoModel;
 import iot.sixiang.license.model.SessionContext;
 import iot.sixiang.license.model.vo.DeviceDetailVo;
 import iot.sixiang.license.model.vo.DeviceVo;
-import iot.sixiang.license.net.TcpServer;
 import iot.sixiang.license.service.DeviceService;
 import iot.sixiang.license.util.CommonUtil;
 import lombok.extern.slf4j.Slf4j;
@@ -15,7 +14,6 @@ import org.springframework.stereotype.Component;
 import javax.annotation.PostConstruct;
 import java.util.*;
 
-
 @Component
 @Slf4j
 public class DeviceManager {
@@ -23,13 +21,15 @@ public class DeviceManager {
 
     private Map<String, SessionContext> sessionContexts = null;
     private DeviceChannelInitializer channelInitializer;
-    private TcpServer server = null;
+//    private TcpServer server = null;
     private int port = 18889;
     private Map<String, DeviceVo> allDevice = null;
     @Autowired
     private DeviceService deviceService;
     @Autowired
     private DeviceServerHandler handler;
+    @Autowired
+    private AsyncTcpServer asyncTcpServer;
 
     public DeviceManager() {
         sessionContexts = new HashMap<String, SessionContext>();
@@ -42,11 +42,11 @@ public class DeviceManager {
         initDevices();
     }
 
+
     private void startTcpService() {
         sessionContexts = new HashMap<String, SessionContext>();
         channelInitializer = new DeviceChannelInitializer(handler);
-        server = new TcpServer(port, channelInitializer);
-        server.start();
+        asyncTcpServer.start(port, channelInitializer);
     }
 
     public void initDevices() {

license/src/main/java/iot/sixiang/license/forward/ForwardClient.java

@@ -33,17 +33,16 @@ public class ForwardClient {
         channelInitializer = new ForwardChannelInitializer(handler);
         EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
         bootstrap = new Bootstrap();
-        if (bootstrap != null) {
-            try {
-                bootstrap
-                        .channel(NioSocketChannel.class)
-                        .option(ChannelOption.SO_KEEPALIVE, true)
-                        .group(eventLoopGroup)
-                        .handler(channelInitializer);
-            } catch (IllegalStateException ex) {
-                log.error(ex.getMessage());
-            }
+        try {
+            bootstrap
+                    .channel(NioSocketChannel.class)
+                    .option(ChannelOption.SO_KEEPALIVE, true)
+                    .group(eventLoopGroup)
+                    .handler(channelInitializer);
+        } catch (IllegalStateException ex) {
+            log.error(ex.getMessage());
         }
+
     }
 
     public void startTcp(String host, int port, String appId) {

license/src/main/java/iot/sixiang/license/net/TcpServer.java

@@ -20,43 +20,38 @@ public class TcpServer {
 	}
 
 	public void start() {
-		Thread thread = new Thread(new Runnable() {
-			@Override
-			public void run() {
-				//创建两个线程组 bossGroup、workerGroup
-				EventLoopGroup bossGroup = new NioEventLoopGroup(4);
-				EventLoopGroup workerGroup = new NioEventLoopGroup(4);
-				log.debug("Tcp服务，开始监听端口:{}",port);
-				//创建服务端的启动对象，设置参数
-				ServerBootstrap b = new ServerBootstrap();
-				//设置两个线程组boosGroup和workerGroup
-				b.group(bossGroup, workerGroup)
-						//设置服务端通道实现类型
-						.channel(NioServerSocketChannel.class)
+
+		//创建两个线程组 bossGroup、workerGroup
+		EventLoopGroup bossGroup = new NioEventLoopGroup(4);
+		EventLoopGroup workerGroup = new NioEventLoopGroup(4);
+		log.debug("Tcp服务，开始监听端口:{}", port);
+		//创建服务端的启动对象，设置参数
+		ServerBootstrap b = new ServerBootstrap();
+		//设置两个线程组boosGroup和workerGroup
+		b.group(bossGroup, workerGroup)
+				//设置服务端通道实现类型
+				.channel(NioServerSocketChannel.class)
 //						.handler(new LoggingHandler(LogLevel.INFO))
-						.childHandler(channelInitializer)
-						// 设置tcp缓冲区
-						.option(ChannelOption.SO_BACKLOG, 1024)
-						//设置保持活动连接状态
-						.childOption(ChannelOption.SO_KEEPALIVE, true);
-				ChannelFuture f;
-				try {
-					f = b.bind(port).sync();
-					f.channel().closeFuture().sync();
-				} catch (InterruptedException e) {
-					// TODO Auto-generated catch block
-					log.error("Tcp服务异常，端口:{}", port);
-				} finally {
-					log.debug("Tcp服务，停止退出");
-					if (workerGroup != null) {
-						workerGroup.shutdownGracefully();
-					}
-					if (bossGroup != null) {
-						bossGroup.shutdownGracefully();
-					}
-				}
+				.childHandler(channelInitializer)
+				// 设置tcp缓冲区
+				.option(ChannelOption.SO_BACKLOG, 1024)
+				//设置保持活动连接状态
+				.childOption(ChannelOption.SO_KEEPALIVE, true);
+		ChannelFuture f;
+		try {
+			f = b.bind(port).sync();
+			f.channel().closeFuture().sync();
+		} catch (InterruptedException e) {
+			// TODO Auto-generated catch block
+			log.error("Tcp服务异常，端口:{}", port);
+		} finally {
+			log.debug("Tcp服务，停止退出");
+			if (workerGroup != null) {
+				workerGroup.shutdownGracefully();
+			}
+			if (bossGroup != null) {
+				bossGroup.shutdownGracefully();
 			}
-		});
-		thread.start();
+		}
 	}
 }