Skip to content

B
byit-myth-job
Commit 200da79c by huangfusuper
Options

修改扫描器锁为redis分布式锁

parent 682d3166
Showing with 636 additions and 200 deletions
byit-myth-admin/src/main/resources/config/application-dev.yml
...@@ -69,3 +69,26 @@ myth: ...@@ -69,3 +69,26 @@ myth:
register: register:
url: ${myth-rpc.registry.address} url: ${myth-rpc.registry.address}
######################哨兵模式#####################
#redisson:
# master-name: myMaster
# sentinel-addresses:
# - 127.0.0.1:3306
# - 127.7.7.7:6379
######################哨兵模式#####################
######################单机环境#####################
redisson:
address: redis://10.0.120.208:6379
database: 3
######################单机环境#####################
######################redis分布式锁#####################
lock:
type: redis
######################redis分布式锁#####################
######################DB行锁#####################
#lock:
# type: db
######################DB行锁#####################
\ No newline at end of file
byit-myth-admin/src/main/resources/config/application-local.yml
...@@ -71,3 +71,27 @@ myth: ...@@ -71,3 +71,27 @@ myth:
biz: ${myth-rpc.registry.biz} biz: ${myth-rpc.registry.biz}
register: register:
url: ${myth-rpc.registry.address} url: ${myth-rpc.registry.address}
######################哨兵模式#####################
#redisson:
# master-name: myMaster
# sentinel-addresses:
# - 127.0.0.1:3306
# - 127.7.7.7:6379
######################哨兵模式#####################
######################单机环境#####################
redisson:
address: redis://10.0.120.208:6379
database: 3
######################单机环境#####################
######################redis分布式锁#####################
lock:
type: redis
######################redis分布式锁#####################
######################DB行锁#####################
#lock:
# type: db
######################DB行锁#####################
\ No newline at end of file
byit-myth-admin/src/main/resources/config/application-test.yml
...@@ -69,3 +69,28 @@ myth: ...@@ -69,3 +69,28 @@ myth:
biz: ${myth-rpc.registry.biz} biz: ${myth-rpc.registry.biz}
register: register:
url: ${myth-rpc.registry.address} url: ${myth-rpc.registry.address}
######################哨兵模式#####################
#redisson:
# master-name: myMaster
# sentinel-addresses:
# - 127.0.0.1:3306
# - 127.7.7.7:6379
######################哨兵模式#####################
######################单机环境#####################
redisson:
address: redis://10.0.120.208:6379
database: 3
######################单机环境#####################
######################redis分布式锁#####################
lock:
type: redis
######################redis分布式锁#####################
######################DB行锁#####################
#lock:
# type: db
######################DB行锁#####################
\ No newline at end of file
byit-myth-core/myth-admin-core/pom.xml
...@@ -87,6 +87,12 @@ ...@@ -87,6 +87,12 @@
<version>1.0-SNAPSHOT</version> <version>1.0-SNAPSHOT</version>
</dependency> </dependency>
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.8.2</version>
</dependency>
</dependencies> </dependencies>
<build> <build>
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobScheduler.java
package com.byit.conf; package com.byit.conf;
import com.byit.factory.DaemonScanThreadRunHelper; import com.byit.factory.DaemonScanThreadRunHelper;
import com.byit.factory.DaemonScanThreadRunHelperDbLock;
import com.byit.thread.BaseThreadRunHelper; import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.DisposableBean;
...@@ -22,11 +23,13 @@ import java.util.Map; ...@@ -22,11 +23,13 @@ import java.util.Map;
public class MythJobScheduler implements InitializingBean, DisposableBean { public class MythJobScheduler implements InitializingBean, DisposableBean {
private final ApplicationContext applicationContext; private final ApplicationContext applicationContext;
private final DaemonScanThreadRunHelper daemonScanThreadRunHelper;
@Autowired @Autowired
public MythJobScheduler(ApplicationContext applicationContext) { public MythJobScheduler(ApplicationContext applicationContext, DaemonScanThreadRunHelper daemonScanThreadRunHelper) {
this.applicationContext = applicationContext; this.applicationContext = applicationContext;
this.daemonScanThreadRunHelper = daemonScanThreadRunHelper;
} }
/** /**
...@@ -34,7 +37,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean { ...@@ -34,7 +37,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
*/ */
@Override @Override
public void destroy() { public void destroy() {
DaemonScanThreadRunHelper.logoutThreadGroup(); daemonScanThreadRunHelper.logoutThreadGroup();
} }
/** /**
...@@ -44,7 +47,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean { ...@@ -44,7 +47,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
public void afterPropertiesSet() { public void afterPropertiesSet() {
helperAutoAdd(); helperAutoAdd();
//启用扫描线程 //启用扫描线程
DaemonScanThreadRunHelper.runDaemonThreads(); daemonScanThreadRunHelper.runDaemonThreads();
} }
/** /**
...@@ -54,7 +57,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean { ...@@ -54,7 +57,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
Map<String, BaseThreadRunHelper> beansOfType = applicationContext.getBeansOfType(BaseThreadRunHelper.class); Map<String, BaseThreadRunHelper> beansOfType = applicationContext.getBeansOfType(BaseThreadRunHelper.class);
beansOfType.forEach((key, value) -> { beansOfType.forEach((key, value) -> {
log.info("-------------自动加载帮助类,{}-------------",key); log.info("-------------自动加载帮助类,{}-------------",key);
DaemonScanThreadRunHelper.addThread(value); daemonScanThreadRunHelper.addThread(value);
}); });
} }
} }
byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/RedissonAutoConfiguration.java 0 → 100644
package com.byit.conf;
import com.byit.conf.properties.RedissonProperties;
import com.byit.util.lock.DistributedLocker;
import com.byit.util.lock.RedissLockUtil;
import com.byit.util.lock.RedissonDistributedLocker;
import org.apache.commons.lang3.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SentinelServersConfig;
import org.redisson.config.SingleServerConfig;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* 分布式锁配置对象
* @author huangfu
*/
@Configuration
@ConditionalOnClass(Config.class)
@EnableConfigurationProperties(RedissonProperties.class)
public class RedissonAutoConfiguration {
@Autowired
private RedissonProperties redssionProperties;
/**
* 哨兵模式自动装配
* @return
*/
@Bean
@ConditionalOnProperty(name="redisson.master-name")
RedissonClient redissonSentinel() {
Config config = new Config();
SentinelServersConfig serverConfig = config.useSentinelServers().addSentinelAddress(redssionProperties.getSentinelAddresses())
.setMasterName(redssionProperties.getMasterName())
.setTimeout(redssionProperties.getTimeout())
.setMasterConnectionPoolSize(redssionProperties.getMasterConnectionPoolSize())
.setSlaveConnectionPoolSize(redssionProperties.getSlaveConnectionPoolSize())
.setDatabase(redssionProperties.getDatabase());
if(StringUtils.isNotBlank(redssionProperties.getPassword())) {
serverConfig.setPassword(redssionProperties.getPassword());
}
return Redisson.create(config);
}
/**
* 单机模式自动装配
* @return
*/
@Bean
@ConditionalOnProperty(name="redisson.address")
RedissonClient redissonSingle() {
Config config = new Config();
SingleServerConfig serverConfig = config.useSingleServer()
.setAddress(redssionProperties.getAddress())
.setTimeout(redssionProperties.getTimeout())
.setConnectionPoolSize(redssionProperties.getConnectionPoolSize())
.setConnectionMinimumIdleSize(redssionProperties.getConnectionMinimumIdleSize())
.setDatabase(redssionProperties.getDatabase());
if(StringUtils.isNotBlank(redssionProperties.getPassword())) {
serverConfig.setPassword(redssionProperties.getPassword());
}
return Redisson.create(config);
}
/**
* 装配locker类，并将实例注入到RedissLockUtil中
* @return
*/
@Bean
DistributedLocker distributedLocker(RedissonClient redissonSingle) {
RedissonDistributedLocker locker = new RedissonDistributedLocker();
locker.setRedissonClient(redissonSingle);
RedissLockUtil.setLocker(locker);
return locker;
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/properties/RedissonProperties.java 0 → 100644
package com.byit.conf.properties;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import java.util.List;
/**
* redisson分布式锁配置类
* @author huangfu
*/
@Data
@ConfigurationProperties(prefix = "redisson")
public class RedissonProperties {
private int timeout = 3000;
private String address;
private String password;
private int connectionPoolSize = 64;
private int connectionMinimumIdleSize=10;
private int slaveConnectionPoolSize = 250;
private int masterConnectionPoolSize = 250;
private String[] sentinelAddresses;
private String masterName;
private int database = 2;
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/BaseDaemonScanThreadRunHelper.java 0 → 100644
package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
* @author huangfu
*/
@Slf4j
public abstract class BaseDaemonScanThreadRunHelper implements DaemonScanThreadRunHelper {
/**
* 控制一组守护线程是否运行
*/
public volatile static boolean THREAD_GROUP_STOP = false;
/**
* 初始睡眠时间 单位毫秒
*/
public static final Long INIT_SLEEP_DATE = 10000L;
/**
* 关闭等待时间 单位毫秒
*/
public static final Long CLOSE_WAIT_TIME = 1000L;
/**
* 循环间隔 单位毫秒
*/
public static final Long CYCLE_INTERVAL = 1000L;
/**
* 线程运行必须原料
*/
public static final Map<String,BaseThreadRunHelper> THREAD_RUN_OPT = new ConcurrentHashMap<>(8);
/**
* 线程存储 用于停止线程
*/
public static final Map<String,Thread> THREADS_MAP = new ConcurrentHashMap<>(8);
/**
* 线程添加 将线程扫描器添加进线程管理池
* @param threadRunHelper 扫描器
*/
@Override
public void addThread(BaseThreadRunHelper threadRunHelper) {
String lockName = threadRunHelper.getLockName();
THREAD_RUN_OPT.put(lockName,threadRunHelper);
}
/**
* 销毁器，将线程管理池里面的扫描器全部注销掉
*/
@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);
});
}
/**
* 对齐时钟。整秒运行
* @param waitTime 休眠时间
* @param threadName 线程名称
*/
public void dateAligned(long waitTime,String threadName){
try {
log.debug("-----------线程{}开始休眠，休眠时间{}---------",threadName,waitTime);
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis() % 1000);
} catch (InterruptedException e) {
log.warn("----------------【{}线程被中断】-----------------------",threadName);
}
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/DaemonScanThreadRunHelper.java
package com.byit.factory; package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper; import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/** /**
* 扫描线程运行帮助 * 线程构建帮助器
* @author huangfu * @author huangfu
*/ */
@Slf4j public interface DaemonScanThreadRunHelper {
public class DaemonScanThreadRunHelper {
/**
* 控制一组守护线程是否运行
*/
private volatile static boolean THREAD_GROUP_STOP = false;
/**
* 初始睡眠时间 单位毫秒
*/
private static final Long INIT_SLEEP_DATE = 10000L;
/**
* 关闭等待时间 单位毫秒
*/
private static final Long CLOSE_WAIT_TIME = 1000L;
/** /**
* 循环间隔 单位毫秒 * 添加线程
* @param threadRunHelper
*/ */
private static final Long CYCLE_INTERVAL = 1000L; void addThread(BaseThreadRunHelper threadRunHelper);
/**
* 线程运行必须原料
*/
private static final Map<String,BaseThreadRunHelper> THREAD_RUN_OPT = new ConcurrentHashMap<>(8);
/**
* 线程存储 用于停止线程
*/
private static final Map<String,Thread> THREADS_MAP = new ConcurrentHashMap<>(8);
/**
* 添加一个线程运行机器
* @param threadRunHelper 线程运行实例
*/
public static void addThread(BaseThreadRunHelper threadRunHelper){
String lockName = threadRunHelper.getLockName();
THREAD_RUN_OPT.put(lockName,threadRunHelper);
}
/** /**
* 运行线程 * 运行线程
*/ */
public static void runDaemonThreads(){ void runDaemonThreads();
Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
entries.forEach(threadExamples ->{
String examplesKey = threadExamples.getKey();
BaseThreadRunHelper examplesValue = threadExamples.getValue();
buildThread(examplesKey,examplesValue);
});
}
/** /**
* 注销线程组 * 注销线程组
*/ */
public static void logoutThreadGroup(){ void logoutThreadGroup();
DaemonScanThreadRunHelper.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);
});
}
/**
* 线程构建
* @param lockName 行锁名称
* @param examplesValue 线程运行资源
*/
private static void buildThread(String lockName, BaseThreadRunHelper examplesValue){
log.info("-----------开始构建扫描线程,线程锁为{}----------------",lockName);
DataSource dataSource = examplesValue.getDataSource();
if(dataSource != null){
String threadClassName = examplesValue.getClass().getSimpleName();
String threadName = "myth-job#【"+threadClassName+"】";
Thread exampleThread = new Thread(() ->{
//初始化睡眠
dateAligned(INIT_SLEEP_DATE,threadName);
log.info("---------------{}线程启动成功----------------",threadName);
while (!THREAD_GROUP_STOP){
dateAligned(CYCLE_INTERVAL,threadName);
//定义睡眠变量
Long sleepTime = 0L;
Connection conn = null;
Boolean connAutoCommit = null;
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.execute();
//调用业务操作
sleepTime = examplesValue.start();
}catch (Exception e){
if(!THREAD_GROUP_STOP){
e.printStackTrace();
}
}finally {
//提交行锁
if (conn != null) {
try {
conn.commit();
} catch (Exception e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
//恢复自动提交
if (conn != null) {
try {
conn.setAutoCommit(connAutoCommit);
} catch (Exception e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
}
//关闭数据库执行器
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
//关闭数据库连接
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
}
sleepTime = sleepTime==null?examplesValue.UNIVERSAL_WAIT_TIME:sleepTime;
dateAligned(sleepTime,threadName);
}
});
exampleThread.setName(threadName);
exampleThread.setDaemon(true);
exampleThread.start();
THREADS_MAP.put(threadName,exampleThread);
}else{
System.out.println("--------------出错了-----------");
}
}
/**
* 对齐时钟。整秒运行
* @param waitTime 休眠时间
* @param threadName 线程名称
*/
private static void dateAligned(long waitTime,String threadName){
try {
log.debug("-----------线程{}开始休眠，休眠时间{}---------",threadName,waitTime);
TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis() % 1000);
} catch (InterruptedException e) {
log.warn("----------------【{}线程被中断】-----------------------",threadName);
}
}
} }
byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/DaemonScanThreadRunHelperDbLock.java 0 → 100644
package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.context.annotation.Primary;
import org.springframework.stereotype.Component;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
/**
* 扫描线程运行帮助 使用数据库锁
* @author huangfu
*/
@Slf4j
@Component
@ConditionalOnExpression("'${lock.type}'.equals('db')")
public class DaemonScanThreadRunHelperDbLock extends BaseDaemonScanThreadRunHelper {
/**
* 运行线程
*/
@Override
public void runDaemonThreads(){
Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
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);
DataSource dataSource = examplesValue.getDataSource();
if(dataSource != null){
String threadClassName = examplesValue.getClass().getSimpleName();
String threadName = "myth-job#【"+threadClassName+"】";
Thread exampleThread = new Thread(() ->{
//初始化睡眠
dateAligned(INIT_SLEEP_DATE,threadName);
log.info("---------------{}线程启动成功----------------",threadName);
while (!THREAD_GROUP_STOP){
dateAligned(CYCLE_INTERVAL,threadName);
//定义睡眠变量
Long sleepTime = 0L;
Connection conn = null;
Boolean connAutoCommit = null;
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.execute();
//调用业务操作
sleepTime = examplesValue.start();
}catch (Exception e){
if(!THREAD_GROUP_STOP){
e.printStackTrace();
}
}finally {
//提交行锁
if (conn != null) {
try {
conn.commit();
} catch (Exception e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【提交行锁出错】---------------------");
}
}
}
//恢复自动提交
if (conn != null) {
try {
conn.setAutoCommit(connAutoCommit);
} catch (Exception e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【恢复自动提交出错】---------------------");
}
}
}
//关闭数据库执行器
if (preparedStatement != null) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
//关闭数据库连接
if (conn != null) {
try {
conn.close();
} catch (SQLException e) {
if (!THREAD_GROUP_STOP) {
log.error("--------------------【关闭执行器出错】---------------------");
}
}
}
}
sleepTime = sleepTime==null?examplesValue.UNIVERSAL_WAIT_TIME:sleepTime;
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 0 → 100644
package com.byit.factory;
import com.byit.thread.BaseThreadRunHelper;
import com.byit.util.lock.RedissLockUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.Set;
/**
* 使用redis分布式锁
* @author huangfu
*/
@Component
@Slf4j
@ConditionalOnExpression("'${lock.type}'.equals('redis')")
public class DaemonScanThreadRunHelperRedisLock extends BaseDaemonScanThreadRunHelper {
@Override
public void runDaemonThreads() {
Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
entries.forEach(threadExamples ->{
String examplesKey = threadExamples.getKey();
BaseThreadRunHelper examplesValue = threadExamples.getValue();
buildThread(examplesKey,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(() ->{
//初始化睡眠
dateAligned(INIT_SLEEP_DATE,threadName);
log.info("---------------{}线程启动成功----------------",threadName);
while (!THREAD_GROUP_STOP){
dateAligned(CYCLE_INTERVAL,threadName);
//定义睡眠变量
Long sleepTime = 0L;
try{
//加锁
RedissLockUtil.lock(lockName,30);
log.info("------------{},加锁成功，锁名称为{}----------",threadName,lockName);
//调用业务操作
sleepTime = examplesValue.start();
}catch (Exception e) {
if(!THREAD_GROUP_STOP){
e.printStackTrace();
}
}finally {
RedissLockUtil.unlock(lockName);
log.info("------------{},解锁成功，锁名称为{}----------",threadName,lockName);
}
sleepTime = sleepTime==null?examplesValue.UNIVERSAL_WAIT_TIME:sleepTime;
dateAligned(sleepTime,threadName);
}
});
exampleThread.setName(threadName);
exampleThread.setDaemon(true);
exampleThread.start();
THREADS_MAP.put(threadName,exampleThread);
}
}
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/EmailThreadRunHelper.java
...@@ -41,7 +41,7 @@ public class EmailThreadRunHelper extends BaseThreadRunHelper { ...@@ -41,7 +41,7 @@ public class EmailThreadRunHelper extends BaseThreadRunHelper {
}else{ }else{
return UNIVERSAL_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
return NOT_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
@Override @Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/EndAndNotWarningThreadRunHelper.java
...@@ -93,7 +93,7 @@ public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper { ...@@ -93,7 +93,7 @@ public class EndAndNotWarningThreadRunHelper extends BaseThreadRunHelper {
}else{ }else{
return UNIVERSAL_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
return NOT_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
@Override @Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/FailedRetryThreadRunHelper.java
...@@ -45,7 +45,7 @@ public class FailedRetryThreadRunHelper extends BaseThreadRunHelper { ...@@ -45,7 +45,7 @@ public class FailedRetryThreadRunHelper extends BaseThreadRunHelper {
}else{ }else{
return UNIVERSAL_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
return NOT_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
@Override @Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/FlowThreadRunHelper.java
...@@ -56,7 +56,7 @@ public class FlowThreadRunHelper extends BaseThreadRunHelper { ...@@ -56,7 +56,7 @@ public class FlowThreadRunHelper extends BaseThreadRunHelper {
}else{ }else{
return UNIVERSAL_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
return NOT_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
@Override @Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/ScheduleThreadRunHelper.java
...@@ -86,7 +86,7 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper { ...@@ -86,7 +86,7 @@ public class ScheduleThreadRunHelper extends BaseThreadRunHelper {
}else{ }else{
return SCHEDULE_READ_MS; return SCHEDULE_READ_MS;
} }
return NOT_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
@Override @Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/TaskThreadRunHelper.java
...@@ -117,11 +117,11 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper { ...@@ -117,11 +117,11 @@ public class TaskThreadRunHelper extends BaseThreadRunHelper {
//执行保存到排表 删除任务表操作 //执行保存到排表 删除任务表操作
taskAndScheduleService.saveScheduleAndDeleteTask(jobTaskSchedules); taskAndScheduleService.saveScheduleAndDeleteTask(jobTaskSchedules);
}else{ }else{
return PRE_READ_MS; return UNIVERSAL_WAIT_TIME;
} }
return NOT_WAIT_TIME; return PRE_READ_MS;
} }
/** /**
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/helper/VirtualNodeThreadRunHelper.java
...@@ -57,7 +57,7 @@ public class VirtualNodeThreadRunHelper extends BaseThreadRunHelper { ...@@ -57,7 +57,7 @@ public class VirtualNodeThreadRunHelper extends BaseThreadRunHelper {
}else{ }else{
return UNIVERSAL_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
return NOT_WAIT_TIME; return UNIVERSAL_WAIT_TIME;
} }
@Override @Override
......
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/lock/DistributedLocker.java 0 → 100644
package com.byit.util.lock;
import java.util.concurrent.TimeUnit;
public interface DistributedLocker {
/**
* 加锁
* @param lockKey
*/
void lock(String lockKey);
/**
* 解锁
* @param lockKey
*/
void unlock(String lockKey);
/**
* 带超时时间的加锁
* @param lockKey
* @param timeout
*/
void lock(String lockKey, int timeout);
/**
* 带单位的超时时间的加锁
* @param lockKey
* @param unit
* @param timeout
*/
void lock(String lockKey, TimeUnit unit , int timeout);
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/lock/RedissLockUtil.java 0 → 100644
package com.byit.util.lock;
import java.util.concurrent.TimeUnit;
/**
* 锁使用帮助器
* @author huangfu
*/
public class RedissLockUtil {
private static DistributedLocker redissLock;
public static void setLocker(DistributedLocker locker) {
redissLock = locker;
}
public static void lock(String lockKey) {
redissLock.lock(lockKey);
}
public static void unlock(String lockKey) {
redissLock.unlock(lockKey);
}
/**
* 带超时的锁
* @param lockKey
* @param timeout 超时时间 单位：秒
*/
public static void lock(String lockKey, int timeout) {
redissLock.lock(lockKey, timeout);
}
/**
* 带超时的锁
* @param lockKey
* @param unit 时间单位
* @param timeout 超时时间
*/
public static void lock(String lockKey, TimeUnit unit , int timeout) {
redissLock.lock(lockKey, unit, timeout);
}
}
\ No newline at end of file
byit-myth-core/myth-admin-core/src/main/java/com/byit/util/lock/RedissonDistributedLocker.java 0 → 100644
package com.byit.util.lock;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import java.util.concurrent.TimeUnit;
/**
* @author huangfu
* redis分布式锁
*/
public class RedissonDistributedLocker implements DistributedLocker {
private RedissonClient redissonClient;
@Override
public void lock(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock();
}
@Override
public void unlock(String lockKey) {
RLock lock = redissonClient.getLock(lockKey);
lock.unlock();
}
@Override
public void lock(String lockKey, int leaseTime) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock(leaseTime, TimeUnit.SECONDS);
}
@Override
public void lock(String lockKey, TimeUnit unit ,int timeout) {
RLock lock = redissonClient.getLock(lockKey);
lock.lock(timeout, unit);
}
public void setRedissonClient(RedissonClient redissonClient) {
this.redissonClient = redissonClient;
}
}
\ No newline at end of file
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