Commit fca591cd by huangfusuper

修改扫描线程为线程池运行

parent 41b9e7a7
package com.byit.factory; package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper; import com.byit.thread.BaseThreadRunHelper;
import com.byit.util.ThreadPoolUtil;
import com.byit.util.lock.RedissLockUtil; import com.byit.util.lock.RedissLockUtil;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
...@@ -8,6 +9,7 @@ import java.util.HashSet; ...@@ -8,6 +9,7 @@ import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
/** /**
...@@ -48,6 +50,12 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR ...@@ -48,6 +50,12 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR
public static final Set<String> LOCK_NAMES = new HashSet<>(8); public static final Set<String> LOCK_NAMES = new HashSet<>(8);
/** /**
* 扫描线程数量
*/
public static ThreadPoolExecutor SCAN_WORLD_THREAD = ThreadPoolUtil.createFixedLengthThreadPoolExecutor("scan flow stream thread",THREAD_RUN_OPT.entrySet().size());
/**
* 线程添加 将线程扫描器添加进线程管理池 * 线程添加 将线程扫描器添加进线程管理池
* @param threadRunHelper 扫描器 * @param threadRunHelper 扫描器
*/ */
...@@ -63,25 +71,27 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR ...@@ -63,25 +71,27 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR
*/ */
@Override @Override
public void logoutThreadGroup() { public void logoutThreadGroup() {
THREAD_GROUP_STOP = true; // THREAD_GROUP_STOP = true;
Set<Map.Entry<String, Thread>> threadExamples = THREADS_MAP.entrySet(); // Set<Map.Entry<String, Thread>> threadExamples = THREADS_MAP.entrySet();
threadExamples.forEach(threadExample ->{ // threadExamples.forEach(threadExample ->{
log.warn("------------开始注销线程{}------------",threadExample); // log.warn("------------开始注销线程{}------------",threadExample);
String threadName = threadExample.getKey(); // String threadName = threadExample.getKey();
Thread thread = threadExample.getValue(); // Thread thread = threadExample.getValue();
dateAligned(CLOSE_WAIT_TIME,threadName); // dateAligned(CLOSE_WAIT_TIME,threadName);
//判断线程是否处于终止状态 // //判断线程是否处于终止状态
if(thread.getState() != Thread.State.TERMINATED){ // if(thread.getState() != Thread.State.TERMINATED){
thread.interrupt(); // thread.interrupt();
//
// try {
// thread.join();
// } catch (InterruptedException e) {
// e.printStackTrace( );
// }
// }
// log.warn("--------{}线程被注销---------",threadName);
// });
try { SCAN_WORLD_THREAD.shutdown();
thread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("--------{}线程被注销---------",threadName);
});
LOCK_NAMES.forEach(lockName ->{ LOCK_NAMES.forEach(lockName ->{
log.warn("--------{}锁消除---------",lockName); log.warn("--------{}锁消除---------",lockName);
......
...@@ -22,11 +22,6 @@ import java.util.concurrent.TimeUnit; ...@@ -22,11 +22,6 @@ import java.util.concurrent.TimeUnit;
@ConditionalOnExpression("'${lock.type}'.equals('redis')") @ConditionalOnExpression("'${lock.type}'.equals('redis')")
public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunHelper { public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunHelper {
/**
* 扫描线程数量
*/
public static ThreadPoolExecutor SCAN_WORLD_THREAD = ThreadPoolUtil.createFixedLengthThreadPoolExecutor("scan flow stream thread",THREAD_RUN_OPT.entrySet().size());
@Override @Override
public void runDaemonThreads() { public void runDaemonThreads() {
Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet(); Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
...@@ -41,7 +36,10 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH ...@@ -41,7 +36,10 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
log.info("-----------开始构建扫描线程,线程锁为{}----------------",lockName); log.info("-----------开始构建扫描线程,线程锁为{}----------------",lockName);
String threadClassName = examplesValue.getClass().getSimpleName(); String threadClassName = examplesValue.getClass().getSimpleName();
String threadName = "myth-job#【"+threadClassName+"】"; String threadName = "myth-job#【"+threadClassName+"】";
Thread exampleThread = new Thread(() ->{ //Thread exampleThread = new Thread();
//exampleThread.setName(threadName);
//exampleThread.setDaemon(true);
SCAN_WORLD_THREAD.execute(() ->{
//初始化睡眠 //初始化睡眠
dateAligned(INIT_SLEEP_DATE,threadName); dateAligned(INIT_SLEEP_DATE,threadName);
log.info("---------------{}线程启动成功----------------",threadName); log.info("---------------{}线程启动成功----------------",threadName);
...@@ -67,10 +65,7 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH ...@@ -67,10 +65,7 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
dateAligned(sleepTime,threadName); dateAligned(sleepTime,threadName);
} }
}); });
//exampleThread.setName(threadName);
//exampleThread.setDaemon(true);
SCAN_WORLD_THREAD.execute(exampleThread);
//exampleThread.start() //exampleThread.start()
THREADS_MAP.put(threadName,exampleThread); //THREADS_MAP.put(threadName,exampleThread);
} }
} }
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment