diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java index eb3449c5..02ebd7aa 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java @@ -88,9 +88,9 @@ public class CanalController { cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID)); ip = getProperty(properties, CanalConstants.CANAL_IP); port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT)); - embededCanalServer = new CanalServerWithEmbedded(); + embededCanalServer = CanalServerWithEmbedded.instance(); embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator - canalServer = new CanalServerWithNetty(embededCanalServer); + canalServer = CanalServerWithNetty.instance(); canalServer.setIp(ip); canalServer.setPort(port); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index e9bb75e2..ab6e9734 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -6,6 +6,7 @@ import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; +import com.alibaba.otter.canal.server.CanalService; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.slf4j.MDC; @@ -37,24 +38,39 @@ import com.google.common.collect.MigrateMap; * @author zebin.xuzb * @version 1.0.0 */ -public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, com.alibaba.otter.canal.server.CanalService { +public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, CanalService { private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class); private Map canalInstances; // private Map lastRollbackPostions; private CanalInstanceGenerator canalInstanceGenerator; + + private static class SingletonHolder { + private static final CanalServerWithEmbedded CANAL_SERVER_WITH_EMBEDDED = new CanalServerWithEmbedded(); + } + + private CanalServerWithEmbedded() { + + } + + public static CanalServerWithEmbedded instance() { + return SingletonHolder.CANAL_SERVER_WITH_EMBEDDED; + } + public void start() { - super.start(); + if (!isStart()) { + super.start(); - canalInstances = MigrateMap.makeComputingMap(new Function() { + canalInstances = MigrateMap.makeComputingMap(new Function() { - public CanalInstance apply(String destination) { - return canalInstanceGenerator.generate(destination); - } - }); + public CanalInstance apply(String destination) { + return canalInstanceGenerator.generate(destination); + } + }); - // lastRollbackPostions = new MapMaker().makeMap(); + // lastRollbackPostions = new MapMaker().makeMap(); + } } public void stop() { diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java index 31fb8153..3df5a15f 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java @@ -33,11 +33,16 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana private Channel serverChannel = null; private ServerBootstrap bootstrap = null; - public CanalServerWithNetty(){ + private static class SingletonHolder { + private static final CanalServerWithNetty CANAL_SERVER_WITH_NETTY = new CanalServerWithNetty(); } - public CanalServerWithNetty(CanalServerWithEmbedded embeddedServer){ - this.embeddedServer = embeddedServer; + private CanalServerWithNetty(){ + this.embeddedServer = CanalServerWithEmbedded.instance(); + } + + public static CanalServerWithNetty instance() { + return SingletonHolder.CANAL_SERVER_WITH_NETTY; } public void start() { @@ -98,8 +103,4 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana this.port = port; } - public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) { - this.embeddedServer = embeddedServer; - } - } diff --git a/server/src/test/java/com/alibaba/otter/canal/server/BaseCanalServerWithEmbededTest.java b/server/src/test/java/com/alibaba/otter/canal/server/BaseCanalServerWithEmbededTest.java index ce6d3b02..10f2f3e5 100644 --- a/server/src/test/java/com/alibaba/otter/canal/server/BaseCanalServerWithEmbededTest.java +++ b/server/src/test/java/com/alibaba/otter/canal/server/BaseCanalServerWithEmbededTest.java @@ -31,7 +31,7 @@ public abstract class BaseCanalServerWithEmbededTest { @Before public void setUp() { - server = new CanalServerWithEmbedded(); + server = CanalServerWithEmbedded.instance(); server.setCanalInstanceGenerator(new CanalInstanceGenerator() { public CanalInstance generate(String destination) { diff --git a/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java b/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java index 3b9cc733..10ca8795 100644 --- a/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java +++ b/server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java @@ -49,7 +49,7 @@ public class CanalServerWithNettyTest { @Before public void setUp() { - CanalServerWithEmbedded embeddedServer = new CanalServerWithEmbedded(); + CanalServerWithEmbedded embeddedServer = CanalServerWithEmbedded.instance(); embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator() { public CanalInstance generate(String destination) {