新增UUID到新同步运行器的Redis配置持久化功能
This commit is contained in:
+4
@@ -116,5 +116,9 @@ public class UuidToNewMonitorServiceImpl implements IUuidToNewMonitorService
|
||||
}
|
||||
properties.setBatchSize(n);
|
||||
}
|
||||
if (syncRunner != null)
|
||||
{
|
||||
syncRunner.persistRuntimeConfig();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,19 +4,32 @@ import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.ApplicationArguments;
|
||||
import org.springframework.boot.ApplicationRunner;
|
||||
import org.springframework.core.annotation.Order;
|
||||
import org.springframework.stereotype.Component;
|
||||
import com.bms.common.core.redis.RedisCache;
|
||||
import com.bms.price.config.UuidToNewProperties;
|
||||
import com.bms.price.migration.service.IUuidToNewSyncService;
|
||||
|
||||
/**
|
||||
* UUID 阴影同步运行器:监控页开关控制;
|
||||
* 开启后立即跑批,一批结束立刻下一批;关闭后当前批结束停。
|
||||
* 开关/线程/批次落 Redis,服务重启后按 Redis 状态恢复。
|
||||
*/
|
||||
@Component
|
||||
public class UuidToNewSyncRunner
|
||||
@Order(100)
|
||||
public class UuidToNewSyncRunner implements ApplicationRunner
|
||||
{
|
||||
private static final Logger log = LoggerFactory.getLogger(UuidToNewSyncRunner.class);
|
||||
|
||||
/** Redis:连续同步开关 */
|
||||
public static final String REDIS_KEY_ENABLED = "bms:mig:uuid-to-new:enabled";
|
||||
/** Redis:线程数 */
|
||||
public static final String REDIS_KEY_THREAD_COUNT = "bms:mig:uuid-to-new:thread-count";
|
||||
/** Redis:批次大小 */
|
||||
public static final String REDIS_KEY_BATCH_SIZE = "bms:mig:uuid-to-new:batch-size";
|
||||
|
||||
private final AtomicBoolean wantRun = new AtomicBoolean(false);
|
||||
private final AtomicBoolean loopActive = new AtomicBoolean(false);
|
||||
|
||||
@@ -26,6 +39,9 @@ public class UuidToNewSyncRunner
|
||||
@Autowired
|
||||
private IUuidToNewSyncService syncService;
|
||||
|
||||
@Autowired
|
||||
private RedisCache redisCache;
|
||||
|
||||
public boolean isEnabled()
|
||||
{
|
||||
return wantRun.get();
|
||||
@@ -38,12 +54,15 @@ public class UuidToNewSyncRunner
|
||||
|
||||
/**
|
||||
* 打开:立即启动连续批循环;关闭:标记停止(当前批跑完)。
|
||||
* 同步落 Redis,供重启恢复。
|
||||
*/
|
||||
public synchronized void setEnabled(boolean enabled)
|
||||
{
|
||||
wantRun.set(enabled);
|
||||
persistEnabled(enabled);
|
||||
if (enabled)
|
||||
{
|
||||
persistRuntimeConfig();
|
||||
startLoopIfNeeded();
|
||||
}
|
||||
else
|
||||
@@ -52,6 +71,129 @@ public class UuidToNewSyncRunner
|
||||
}
|
||||
}
|
||||
|
||||
/** 将当前线程/批次写入 Redis(无过期) */
|
||||
public void persistRuntimeConfig()
|
||||
{
|
||||
if (redisCache == null || properties == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
redisCache.setCacheObject(REDIS_KEY_THREAD_COUNT, Integer.valueOf(properties.resolvedThreadCount()));
|
||||
redisCache.setCacheObject(REDIS_KEY_BATCH_SIZE, Integer.valueOf(properties.resolvedBatchSize()));
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
log.warn("uuid-to-new persist runtime config to redis failed: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private void persistEnabled(boolean enabled)
|
||||
{
|
||||
if (redisCache == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
redisCache.setCacheObject(REDIS_KEY_ENABLED, Boolean.valueOf(enabled));
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
log.warn("uuid-to-new persist enabled={} to redis failed: {}", enabled, e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void run(ApplicationArguments args)
|
||||
{
|
||||
restoreFromRedis();
|
||||
}
|
||||
|
||||
/** 启动时从 Redis 恢复线程/批次与开关 */
|
||||
private void restoreFromRedis()
|
||||
{
|
||||
if (redisCache == null)
|
||||
{
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
Object threadObj = redisCache.getCacheObject(REDIS_KEY_THREAD_COUNT);
|
||||
Object batchObj = redisCache.getCacheObject(REDIS_KEY_BATCH_SIZE);
|
||||
if (properties != null)
|
||||
{
|
||||
Integer threads = toInteger(threadObj);
|
||||
if (threads != null)
|
||||
{
|
||||
properties.setThreadCount(threads.intValue());
|
||||
}
|
||||
Integer batch = toInteger(batchObj);
|
||||
if (batch != null)
|
||||
{
|
||||
properties.setBatchSize(batch.intValue());
|
||||
}
|
||||
}
|
||||
Object enabledObj = redisCache.getCacheObject(REDIS_KEY_ENABLED);
|
||||
boolean on = toBoolean(enabledObj);
|
||||
if (on)
|
||||
{
|
||||
log.info("uuid-to-new restore from redis: enabled=true, threads={}, batch={}",
|
||||
properties == null ? null : properties.resolvedThreadCount(),
|
||||
properties == null ? null : properties.resolvedBatchSize());
|
||||
wantRun.set(true);
|
||||
startLoopIfNeeded();
|
||||
}
|
||||
else
|
||||
{
|
||||
log.info("uuid-to-new restore from redis: enabled=false");
|
||||
}
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
log.warn("uuid-to-new restore from redis failed: {}", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
private static boolean toBoolean(Object v)
|
||||
{
|
||||
if (v == null)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
if (v instanceof Boolean)
|
||||
{
|
||||
return ((Boolean) v).booleanValue();
|
||||
}
|
||||
if (v instanceof Number)
|
||||
{
|
||||
return ((Number) v).intValue() != 0;
|
||||
}
|
||||
return "true".equalsIgnoreCase(String.valueOf(v).trim())
|
||||
|| "1".equals(String.valueOf(v).trim());
|
||||
}
|
||||
|
||||
private static Integer toInteger(Object v)
|
||||
{
|
||||
if (v == null)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
if (v instanceof Number)
|
||||
{
|
||||
return Integer.valueOf(((Number) v).intValue());
|
||||
}
|
||||
try
|
||||
{
|
||||
return Integer.valueOf(String.valueOf(v).trim());
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
private void startLoopIfNeeded()
|
||||
{
|
||||
if (!loopActive.compareAndSet(false, true))
|
||||
|
||||
Reference in New Issue
Block a user