|
|
@@ -1,7 +1,7 @@
|
|
|
package com.its.common.cluster.service;
|
|
|
|
|
|
import com.its.common.cluster.vo.*;
|
|
|
-import com.its.common.cluster.utils.HaUtils;
|
|
|
+import com.its.common.cluster.utils.ClusterUtils;
|
|
|
import io.netty.channel.Channel;
|
|
|
import io.netty.channel.ChannelFuture;
|
|
|
import lombok.RequiredArgsConstructor;
|
|
|
@@ -19,21 +19,21 @@ import java.util.concurrent.ScheduledFuture;
|
|
|
|
|
|
@Slf4j
|
|
|
@RequiredArgsConstructor
|
|
|
-public abstract class AbstractHaClusterSlaveService {
|
|
|
+public abstract class AbstractClusterSlaveService {
|
|
|
|
|
|
private final ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
|
|
|
private final ExecutorService executorService= Executors.newFixedThreadPool(1);
|
|
|
- private final List<HaClusterSlave> clientTasks = Collections.synchronizedList(new ArrayList<>());
|
|
|
+ private final List<ClusterSlave> clientTasks = Collections.synchronizedList(new ArrayList<>());
|
|
|
|
|
|
- private final AbstractHaClusterConfig clusterConfig;
|
|
|
+ private final AbstractClusterConfig clusterConfig;
|
|
|
|
|
|
private ScheduledFuture<?> taskFuture;
|
|
|
- private HaClusterSlaveBootstrapFactory bootstrapFactory;
|
|
|
+ private ClusterSlaveBootstrapFactory bootstrapFactory;
|
|
|
|
|
|
|
|
|
@PostConstruct
|
|
|
void init() {
|
|
|
- this.bootstrapFactory = new HaClusterSlaveBootstrapFactory(1, 5);
|
|
|
+ this.bootstrapFactory = new ClusterSlaveBootstrapFactory(1, 5);
|
|
|
this.taskScheduler.setPoolSize(1);
|
|
|
this.taskScheduler.initialize();
|
|
|
}
|
|
|
@@ -44,12 +44,12 @@ public abstract class AbstractHaClusterSlaveService {
|
|
|
}
|
|
|
|
|
|
// 모든 HaCluster 접속
|
|
|
- for (Map.Entry<Integer, HaInfo> entry : this.clusterConfig.getClusterMap().entrySet()) {
|
|
|
- HaInfo cluster = entry.getValue();
|
|
|
- if (cluster.getServerId() == this.clusterConfig.getServerId()) {
|
|
|
+ for (Map.Entry<Integer, ClusterNode> entry : this.clusterConfig.getClusterMap().entrySet()) {
|
|
|
+ ClusterNode cluster = entry.getValue();
|
|
|
+ if (cluster.getId() == this.clusterConfig.getId()) {
|
|
|
continue;
|
|
|
}
|
|
|
- HaClusterSlave slaveClient = new HaClusterSlave(this, clusterConfig, cluster, this.bootstrapFactory);
|
|
|
+ ClusterSlave slaveClient = new ClusterSlave(this, clusterConfig, cluster, this.bootstrapFactory);
|
|
|
this.clientTasks.add(slaveClient);
|
|
|
}
|
|
|
|
|
|
@@ -76,9 +76,9 @@ public abstract class AbstractHaClusterSlaveService {
|
|
|
}
|
|
|
this.taskScheduler.shutdown();
|
|
|
|
|
|
- for (Map.Entry<Integer, HaInfo> entry : this.clusterConfig.getClusterMap().entrySet()) {
|
|
|
- HaInfo cluster = entry.getValue();
|
|
|
- if (cluster.getServerId() == this.clusterConfig.getServerId()) {
|
|
|
+ for (Map.Entry<Integer, ClusterNode> entry : this.clusterConfig.getClusterMap().entrySet()) {
|
|
|
+ ClusterNode cluster = entry.getValue();
|
|
|
+ if (cluster.getId() == this.clusterConfig.getId()) {
|
|
|
continue;
|
|
|
}
|
|
|
channelClose(cluster.getSyncState().getChannel());
|
|
|
@@ -108,12 +108,12 @@ public abstract class AbstractHaClusterSlaveService {
|
|
|
return sdfDate.format(dtLog);
|
|
|
}
|
|
|
|
|
|
- public abstract List<HaClusterMessageData> getClusterMessageData();
|
|
|
+ public abstract List<ClusterMessageData> getClusterMessageData();
|
|
|
|
|
|
- private HaClusterMessage getClusterMessage() {
|
|
|
- List<HaClusterMessageData> details = getClusterMessageData();
|
|
|
- return HaClusterMessage.builder()
|
|
|
- .serverId(this.clusterConfig.getServerId())
|
|
|
+ private ClusterMessage getClusterMessage() {
|
|
|
+ List<ClusterMessageData> details = getClusterMessageData();
|
|
|
+ return ClusterMessage.builder()
|
|
|
+ .nodeId(this.clusterConfig.getId())
|
|
|
.master(this.clusterConfig.isMaster())
|
|
|
.serverTime(getSysTime())
|
|
|
.infos(details)
|
|
|
@@ -133,12 +133,12 @@ public abstract class AbstractHaClusterSlaveService {
|
|
|
log.info("ClusterSlaveService:dataSyncSchedule: {} seconds.", this.clusterConfig.getSyncSeconds());
|
|
|
}
|
|
|
this.taskFuture = this.taskScheduler.scheduleAtFixedRate(() -> {
|
|
|
- for (Map.Entry<Integer, HaInfo> entry : this.clusterConfig.getClusterMap().entrySet()) {
|
|
|
- HaInfo cluster = entry.getValue();
|
|
|
- if (cluster.getServerId() == this.clusterConfig.getServerId()) {
|
|
|
+ for (Map.Entry<Integer, ClusterNode> entry : this.clusterConfig.getClusterMap().entrySet()) {
|
|
|
+ ClusterNode cluster = entry.getValue();
|
|
|
+ if (cluster.getId() == this.clusterConfig.getId()) {
|
|
|
continue;
|
|
|
}
|
|
|
- if (cluster.getSyncState().getState() != HaNET.CLOSED) {
|
|
|
+ if (cluster.getSyncState().getState() != ClusterNET.CLOSED) {
|
|
|
sendSyncData(cluster, cluster.getSyncState().getChannel());
|
|
|
}
|
|
|
}
|
|
|
@@ -147,8 +147,8 @@ public abstract class AbstractHaClusterSlaveService {
|
|
|
|
|
|
}
|
|
|
|
|
|
- public void sendSyncData(final HaInfo cluster, final Channel channel) {
|
|
|
- HaClusterMessage clusterMsg = getClusterMessage();
|
|
|
+ public void sendSyncData(final ClusterNode cluster, final Channel channel) {
|
|
|
+ ClusterMessage clusterMsg = getClusterMessage();
|
|
|
|
|
|
if (this.clusterConfig.isLogging()) {
|
|
|
MDC.put("id", cluster.getLogKey());
|
|
|
@@ -159,18 +159,18 @@ public abstract class AbstractHaClusterSlaveService {
|
|
|
f.awaitUninterruptibly();
|
|
|
if (f.isDone() || f.isSuccess()) {
|
|
|
if (this.clusterConfig.isLogging()) {
|
|
|
- log.info("ClusterSlaveService.sendSyncData: [{}], {}, [--TO: serverId: {}, (serverId: {}, serverTime: {})]",
|
|
|
- this.clusterConfig.getServerId(), HaUtils.getTcpAddress(channel),
|
|
|
- cluster.getServerId(), clusterMsg.getServerId(), clusterMsg.getServerTime());
|
|
|
+ log.info("ClusterSlaveService.sendSyncData: [{}], {}, [--TO: nodeId: {}, (nodeId: {}, serverTime: {})]",
|
|
|
+ this.clusterConfig.getId(), ClusterUtils.getTcpAddress(channel),
|
|
|
+ cluster.getId(), clusterMsg.getNodeId(), clusterMsg.getServerTime());
|
|
|
}
|
|
|
}
|
|
|
}
|
|
|
catch (Exception e) {
|
|
|
- log.error("ClusterSlaveService.sendSyncData: [{}], {}, Failed: [--TO: serverId: {}, (serverId: {}, serverTime: {})]",
|
|
|
- this.clusterConfig.getServerId(), HaUtils.getTcpAddress(channel),
|
|
|
- cluster.getServerId(), clusterMsg.getServerId(), clusterMsg.getServerTime());
|
|
|
+ log.error("ClusterSlaveService.sendSyncData: [{}], {}, Failed: [--TO: nodeId: {}, (nodeId: {}, serverTime: {})]",
|
|
|
+ this.clusterConfig.getId(), ClusterUtils.getTcpAddress(channel),
|
|
|
+ cluster.getId(), clusterMsg.getNodeId(), clusterMsg.getServerTime());
|
|
|
log.error("ClusterSlaveService.sendSyncData: [{}], {}, Failed: {}",
|
|
|
- this.clusterConfig.getServerId(), HaUtils.getTcpAddress(channel), e.getMessage());
|
|
|
+ this.clusterConfig.getId(), ClusterUtils.getTcpAddress(channel), e.getMessage());
|
|
|
}
|
|
|
|
|
|
if (this.clusterConfig.isLogging()) {
|