Skip to content

B
byit-myth-job
Commit 0b84f54e by huangfusuper
Options

修改线程工厂的实现逻辑

parent b88c628e
Showing with 59 additions and 59 deletions
byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/BaseDaemonScanThreadRunHelper.java
package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper;
import com.byit.util.ThreadPoolUtil;
import com.byit.util.lock.RedissLockUtil;
import lombok.extern.slf4j.Slf4j;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
/**
* @author huangfu
*/
@Slf4j
public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadRunHelper {
/**
* 线程池
*/
public static ThreadPoolExecutor SCAN_WORLD_THREAD = null;
/**
* 控制一组守护线程是否运行
*/
......@@ -26,10 +28,7 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR
* 初始睡眠时间 单位毫秒
*/
public static final Long INIT_SLEEP_DATE = 10000L;
/**
* 关闭等待时间 单位毫秒
*/
public static final Long CLOSE_WAIT_TIME = 1000L;
/**
* 循环间隔 单位毫秒
*/
......@@ -39,10 +38,6 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR
* 线程运行必须原料
*/
public static final Map<String,BaseThreadRunHelper> THREAD_RUN_OPT = new ConcurrentHashMap<>(8);
/**
* 线程存储 用于停止线程
*/
public static final Map<String,Thread> THREADS_MAP = new ConcurrentHashMap<>(8);
/**
* 用于磁存储所有的锁
......@@ -68,25 +63,7 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR
@Override
public void logoutThreadGroup() {
THREAD_GROUP_STOP = true;
Set<Map.Entry<String, Thread>> threadExamples = THREADS_MAP.entrySet();
threadExamples.forEach(threadExample ->{
log.warn("------------开始注销线程{}------------",threadExample);
String threadName = threadExample.getKey();
Thread thread = threadExample.getValue();
dateAligned(CLOSE_WAIT_TIME,threadName);
//判断线程是否处于终止状态
if(thread.getState() != Thread.State.TERMINATED){
thread.interrupt();
try {
thread.join();
} catch (InterruptedException e) {
e.printStackTrace( );
}
}
log.warn("--------{}线程被注销---------",threadName);
});
SCAN_WORLD_THREAD.shutdown();
LOCK_NAMES.forEach(lockName ->{
log.warn("--------{}锁消除---------",lockName);
......@@ -107,4 +84,17 @@ public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadR
log.warn("----------------【{}线程被中断】-----------------------",threadName);
}
}
/**
* 通过该线程池创建的守护 全部设置为 设置为守护线程
* @return 对应的线程工厂
*/
public ThreadFactory buildThreadFactory(){
return r -> {
Thread thread = Executors.defaultThreadFactory().newThread(r);
thread.setDaemon(true);
thread.setName(String.format("[byit-myth-job]-scan flow stream thread-Thread:%s",thread.hashCode()));
return thread;
};
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/DaemonScanThreadRunHelperDbLock.java
package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper;
import com.byit.job.utils.MythLogUtils;
import com.byit.thread.BaseThreadRunHelper;
import com.byit.util.ThreadPoolUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
......@@ -27,6 +28,7 @@ public class DaemonScanThreadRunHelperDbLock extends BaseDaemonScanThreadRunHelp
@Override
public void runDaemonThreads(){
Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
SCAN_WORLD_THREAD = ThreadPoolUtil.createFixedLengthThreadPoolExecutor(buildThreadFactory(),entries.size());
entries.forEach(threadExamples ->{
String examplesKey = threadExamples.getKey();
BaseThreadRunHelper examplesValue = threadExamples.getValue();
......@@ -45,7 +47,8 @@ public class DaemonScanThreadRunHelperDbLock extends BaseDaemonScanThreadRunHelp
if(dataSource != null){
String threadClassName = examplesValue.getClass().getSimpleName();
String threadName = "myth-job#【"+threadClassName+"】";
Thread exampleThread = new Thread(() ->{
SCAN_WORLD_THREAD.execute(() ->{
//初始化睡眠
dateAligned(INIT_SLEEP_DATE,threadName);
log.info("---------------{}线程启动成功----------------",threadName);
......@@ -54,16 +57,14 @@ public class DaemonScanThreadRunHelperDbLock extends BaseDaemonScanThreadRunHelp
//定义睡眠变量
Long sleepTime = 0L;
Connection conn = null;
Boolean connAutoCommit = null;
boolean connAutoCommit = true;
PreparedStatement preparedStatement = null;
try {
/**
* 添加行锁
*/
//添加行锁
conn = dataSource.getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
preparedStatement = conn.prepareStatement("SELECT * FROM job_lock WHERE LOCK_NAME = '"+lockName+"' FOR UPDATE ");
preparedStatement = conn.prepareStatement(String.format("SELECT * FROM job_lock WHERE LOCK_NAME = '%s' FOR UPDATE",lockName));
preparedStatement.execute();
//调用业务操作
sleepTime = examplesValue.start();
......@@ -118,11 +119,6 @@ public class DaemonScanThreadRunHelperDbLock extends BaseDaemonScanThreadRunHelp
dateAligned(sleepTime,threadName);
}
});
exampleThread.setName(threadName);
exampleThread.setDaemon(true);
exampleThread.start();
THREADS_MAP.put(threadName,exampleThread);
}else{
System.out.println("--------------出错了-----------");
}
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/DaemonScanThreadRunHelperRedisLock.java
package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper;
import com.byit.job.utils.MythLogUtils;
import com.byit.thread.BaseThreadRunHelper;
import com.byit.util.ThreadPoolUtil;
import com.byit.util.lock.RedissLockUtil;
import lombok.extern.slf4j.Slf4j;
......@@ -10,7 +10,6 @@ import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
......@@ -22,27 +21,29 @@ import java.util.concurrent.TimeUnit;
@ConditionalOnExpression("'${lock.type}'.equals('redis')")
public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunHelper {
/**
* 扫描线程数量
*/
public static ThreadPoolExecutor SCAN_WORLD_THREAD = null;
@Override
public void runDaemonThreads() {
Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
SCAN_WORLD_THREAD = ThreadPoolUtil.createFixedLengthThreadPoolExecutor("scan flow stream thread",entries.size());
SCAN_WORLD_THREAD = ThreadPoolUtil.createFixedLengthThreadPoolExecutor(buildThreadFactory(),entries.size());
entries.forEach(threadExamples ->{
String examplesKey = threadExamples.getKey();
BaseThreadRunHelper examplesValue = threadExamples.getValue();
buildThread(examplesKey,examplesValue);
});
}
/**
* 构建一条线程
* @param lockName 所名称
* @param examplesValue 锁构建的值
*/
private void buildThread(String lockName, BaseThreadRunHelper examplesValue){
log.info("-----------开始构建扫描线程,线程锁为{}----------------",lockName);
String threadClassName = examplesValue.getClass().getSimpleName();
String threadName = "myth-job#【"+threadClassName+"】";
Thread exampleThread = new Thread(() ->{
SCAN_WORLD_THREAD.execute(() ->{
//初始化睡眠
dateAligned(INIT_SLEEP_DATE,threadName);
log.info("---------------{}线程启动成功----------------",threadName);
......@@ -50,8 +51,9 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
//定义睡眠变量
Long sleepTime = 0L;
try{
log.debug("---{}准备加锁---",threadName);
//加锁 60秒后超时
if (RedissLockUtil.trlock(lockName, TimeUnit.SECONDS, 59)) {
if (RedissLockUtil.trlock(lockName, TimeUnit.SECONDS, -1)) {
log.debug("------------{},加锁成功，锁名称为{}----------",threadName,lockName);
dateAligned(CYCLE_INTERVAL,threadName);
//调用业务操作
......@@ -68,10 +70,5 @@ public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunH
dateAligned(sleepTime,threadName);
}
});
exampleThread.setName(threadName);
exampleThread.setDaemon(true);
SCAN_WORLD_THREAD.execute(exampleThread);
//exampleThread.start();
THREADS_MAP.put(threadName,exampleThread);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/BaseThreadRunHelper.java
......@@ -24,7 +24,9 @@ public abstract class BaseThreadRunHelper {
* 返回数据库运行实例
* @return 数据路连接
*/
public abstract DataSource getDataSource();
public DataSource getDataSource(){
return null;
}
/**
* 返回行锁名称
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/ThreadPoolUtil.java
......@@ -2,7 +2,10 @@ package com.byit.util;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.util.concurrent.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 线程池配置
......@@ -41,6 +44,18 @@ public class ThreadPoolUtil {
/**
* 创建定长线程池。特性如下： 可以自定义线程工厂
* @param threadCount 线程数量
* @return 线程池
*/
public static ThreadPoolExecutor createFixedLengthThreadPoolExecutor(ThreadFactory threadFactory, int threadCount) {
return new ThreadPoolExecutor(threadCount, threadCount,
0L, TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(1024), threadFactory, new ThreadPoolExecutor.AbortPolicy());
}
/**
* 定制化线程池
* @param threadName 线程名称
* @param coreCount 核心线程数量
......
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