|
|
@@ -43,13 +43,17 @@ public class HaClusterSlave implements Callable<Object> {
|
|
|
public Object call() {
|
|
|
|
|
|
try {
|
|
|
- MDC.put("id", this.cluster.getLogKey());
|
|
|
+ if (this.clusterConfig.isLogging()) {
|
|
|
+ MDC.put("id", this.cluster.getLogKey());
|
|
|
+ }
|
|
|
|
|
|
this.ipAddress = this.cluster.getIpAddress();
|
|
|
this.port = this.cluster.getSyncPort();
|
|
|
|
|
|
if (this.bootstrap == null) {
|
|
|
- log.info("ClusterSlave >>>>>>>>Start: [{}, {}], {}", this.cluster.getServerId(), this.ipAddress, this.port);
|
|
|
+ if (this.clusterConfig.isLogging()) {
|
|
|
+ log.info("HaClusterSlave: >>>>>>>>Start: [{}, {}], {}", this.cluster.getServerId(), this.ipAddress, this.port);
|
|
|
+ }
|
|
|
this.bootstrap = this.bootstrapFactory.createBootstrap();
|
|
|
this.bootstrap.option(ChannelOption.SO_REUSEADDR, true);
|
|
|
this.bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5 * 1000);
|
|
|
@@ -57,23 +61,26 @@ public class HaClusterSlave implements Callable<Object> {
|
|
|
// 핸들러가 실행되는 순서는 추가된 순서에 의해 결정된다.(Inbound: head=>tail, Outbound: tail=>head, name2ctx)
|
|
|
@Override
|
|
|
public void initChannel(SocketChannel ch) {
|
|
|
-// if (cluster.isLogging()) {
|
|
|
-// ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));
|
|
|
-// }
|
|
|
- IdleStateHandler idleStateHandler = new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS);
|
|
|
-
|
|
|
- ch.pipeline().addLast(idleStateHandler);
|
|
|
- ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(8192, 0, 4));
|
|
|
- ch.pipeline().addLast(new HaClusterMessageDecoder(clusterConfig.isLogging()));
|
|
|
- ch.pipeline().addLast(new HaClusterMessageEncoder(clusterConfig.isLogging()));
|
|
|
- ch.pipeline().addLast(new HaClusterSlaveHandler(slaveService, cluster));
|
|
|
+ if (clusterConfig.isPacketLogging()) {
|
|
|
+ ch.pipeline().addLast(new LoggingHandler(LogLevel.INFO));
|
|
|
+ }
|
|
|
+ ch.pipeline().addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS));
|
|
|
+ ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(8192, 0, 4));
|
|
|
+ ch.pipeline().addLast(new HaClusterMessageDecoder(clusterConfig.isLogging()));
|
|
|
+ ch.pipeline().addLast(new HaClusterMessageEncoder(clusterConfig.isLogging()));
|
|
|
+ ch.pipeline().addLast(new HaClusterSlaveHandler(slaveService, cluster));
|
|
|
}
|
|
|
});
|
|
|
- // 바인드 로컬 포트 설정
|
|
|
+
|
|
|
+ // 바인딩 로컬 포트 설정(설정파일에 등록된 바인딩 포트번호 + 서버 ID), 따라서 하나의 서버에 여러 개의 클러스터 서버가 존재할 수 있다.
|
|
|
+ // 주의) 하나의 서버에 여러 개의 클러스터 서버가 존재할 경우, 바인딩 포트는 위의 규칙에 속하지 않는 포트번호를 할당해야 한다.
|
|
|
this.bootstrap.localAddress(new InetSocketAddress(this.port + this.clusterConfig.getServerId()));
|
|
|
}
|
|
|
|
|
|
- log.info("ClusterSlave >>Connect Try: [{}, {}], {}", this.cluster.getServerId(), this.ipAddress, this.port);
|
|
|
+ if (this.clusterConfig.isLogging()) {
|
|
|
+ log.info("HaClusterSlave: >>Connect Try: [{}, {}], {}", this.cluster.getServerId(), this.ipAddress, this.port);
|
|
|
+ }
|
|
|
+
|
|
|
if (this.channelFuture != null && this.channelFuture.channel() != null) {
|
|
|
this.channelFuture.channel().close();
|
|
|
this.channelFuture = null;
|
|
|
@@ -107,19 +114,24 @@ public class HaClusterSlave implements Callable<Object> {
|
|
|
* 연결 성공시 처리 이벤트
|
|
|
*/
|
|
|
protected void channelOpen(ChannelFuture future) {
|
|
|
- try {
|
|
|
+ if (this.clusterConfig.isLogging()) {
|
|
|
MDC.put("id", this.cluster.getLogKey());
|
|
|
- if (future.isSuccess()) {
|
|
|
- Channel channel = future.channel();
|
|
|
- log.info("ClusterSlave ..channelOpen: [{}, {}], {}, Channel: {}", this.cluster.getServerId(), this.ipAddress, this.port, channel);
|
|
|
- channel.attr(AbstractHaClusterConfig.CLUSTER_ATTRIBUTE_KEY).set(this.cluster);
|
|
|
- this.cluster.getSyncState().connect(channel);
|
|
|
- }
|
|
|
- else {
|
|
|
- log.warn("ClusterSlave ConnectFailed: [{}, {}], {}, Cause: {}", cluster.getServerId(), cluster.getIpAddress(), cluster.getSyncPort(), future.cause().getMessage());
|
|
|
+ }
|
|
|
+ if (future.isSuccess()) {
|
|
|
+ Channel channel = future.channel();
|
|
|
+
|
|
|
+ if (this.clusterConfig.isLogging()) {
|
|
|
+ log.info("HaClusterSlave: channelOpen: [{}, {}], {}, Channel: {}", this.cluster.getServerId(), this.ipAddress, this.port, channel);
|
|
|
}
|
|
|
+
|
|
|
+ channel.attr(AbstractHaClusterConfig.CLUSTER_ATTRIBUTE_KEY).set(this.cluster);
|
|
|
+ this.cluster.getSyncState().connect(channel);
|
|
|
}
|
|
|
- finally {
|
|
|
+ else {
|
|
|
+ log.warn("HaClusterSlave: ConnectFailed: [{}, {}], {}, Cause: {}", cluster.getServerId(), cluster.getIpAddress(), cluster.getSyncPort(), future.cause().getMessage());
|
|
|
+ }
|
|
|
+
|
|
|
+ if (this.clusterConfig.isLogging()) {
|
|
|
MDC.remove(this.cluster.getLogKey());
|
|
|
MDC.clear();
|
|
|
}
|
|
|
@@ -130,21 +142,19 @@ public class HaClusterSlave implements Callable<Object> {
|
|
|
* @param future
|
|
|
*/
|
|
|
protected synchronized void channelClosed(ChannelFuture future) {
|
|
|
- try {
|
|
|
- MDC.put("id", this.cluster.getLogKey());
|
|
|
+ Channel channel = future.channel();
|
|
|
|
|
|
- Channel channel = future.channel();
|
|
|
- log.warn("ClusterSlave channelClosed: [{}, {}], {}, Channel: {}", this.cluster.getServerId(), this.ipAddress, this.port, channel);
|
|
|
-
|
|
|
- channel.attr(AbstractHaClusterConfig.CLUSTER_ATTRIBUTE_KEY).set(null);
|
|
|
- this.cluster.getSyncState().disConnect();
|
|
|
- channel.close();
|
|
|
- channel.eventLoop().schedule(this, 5, TimeUnit.SECONDS);
|
|
|
- }
|
|
|
- finally {
|
|
|
+ if (this.clusterConfig.isLogging()) {
|
|
|
+ MDC.put("id", this.cluster.getLogKey());
|
|
|
+ log.warn("ClusterSlave:channelClosed: [{}, {}], {}, Channel: {}", this.cluster.getServerId(), this.ipAddress, this.port, channel);
|
|
|
MDC.remove(this.cluster.getLogKey());
|
|
|
MDC.clear();
|
|
|
}
|
|
|
+
|
|
|
+ channel.attr(AbstractHaClusterConfig.CLUSTER_ATTRIBUTE_KEY).set(null);
|
|
|
+ this.cluster.getSyncState().disConnect();
|
|
|
+ channel.close();
|
|
|
+ channel.eventLoop().schedule(this, 5, TimeUnit.SECONDS);
|
|
|
}
|
|
|
|
|
|
}
|
HANTE