mirror of
https://github.com/alibaba/canal.git
synced 2026-11-04 00:30:15 +08:00
make canal server become a single instance
This commit is contained in:
@@ -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);
|
||||
|
||||
|
||||
@@ -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<String, CanalInstance> canalInstances;
|
||||
// private Map<ClientIdentity, Position> 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<String, CanalInstance>() {
|
||||
canalInstances = MigrateMap.makeComputingMap(new Function<String, CanalInstance>() {
|
||||
|
||||
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() {
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user