chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #115 from wenerme/master

Browse Source 
Typo,javadoc 修正, 更新 guava 到18.0
This commit is contained in:
agapple
2015-04-28 20:26:41 +08:00
parent ae521195c5 596b060649
commit ea889cb004
39 changed files with 797 additions and 620 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java
View File

@@ -19,14 +19,14 @@ public interface CanalConnector {
* *
* @throws CanalClientException * @throws CanalClientException
*/ */
public void connect() throws CanalClientException; void connect() throws CanalClientException;
/** /**
* 释放链接 * 释放链接
* *
* @throws CanalClientException * @throws CanalClientException
*/ */
public void disconnect() throws CanalClientException; void disconnect() throws CanalClientException;
/** /**
* 检查下链接是否合法 * 检查下链接是否合法
@@ -43,7 +43,7 @@ public interface CanalConnector {
* *
* @throws CanalClientException * @throws CanalClientException
*/ */
public boolean checkValid() throws CanalClientException; boolean checkValid() throws CanalClientException;
/** /**
* 客户端订阅，重复订阅时会更新对应的filter信息 * 客户端订阅，重复订阅时会更新对应的filter信息
@@ -56,7 +56,6 @@ public interface CanalConnector {
* TODO: 后续可以考虑，如果本次提交的filter不为空，在执行过滤时，是对canal server filter + 本次filter的交集处理，达到只取1份binlog数据，多个客户端消费不同的表 * TODO: 后续可以考虑，如果本次提交的filter不为空，在执行过滤时，是对canal server filter + 本次filter的交集处理，达到只取1份binlog数据，多个客户端消费不同的表
* </pre> * </pre>
* *
* @param clientIdentity
* @throws CanalClientException * @throws CanalClientException
*/ */
void subscribe(String filter) throws CanalClientException; void subscribe(String filter) throws CanalClientException;
@@ -64,7 +63,6 @@ public interface CanalConnector {
/** /**
* 客户端订阅，不提交客户端filter，以服务端的filter为准 * 客户端订阅，不提交客户端filter，以服务端的filter为准
* *
* @param clientIdentity
* @throws CanalClientException * @throws CanalClientException
*/ */
void subscribe() throws CanalClientException; void subscribe() throws CanalClientException;
@@ -72,7 +70,6 @@ public interface CanalConnector {
/** /**
* 取消订阅 * 取消订阅
* *
* @param clientIdentity
* @throws CanalClientException * @throws CanalClientException
*/ */
void unsubscribe() throws CanalClientException; void unsubscribe() throws CanalClientException;
@@ -140,14 +137,14 @@ public interface CanalConnector {
void ack(long batchId) throws CanalClientException; void ack(long batchId) throws CanalClientException;
/** /**
* 回滚到未进行 {@link ack} 的地方，指定回滚具体的batchId * 回滚到未进行 {@link #ack} 的地方，指定回滚具体的batchId
* *
* @throws CanalClientException * @throws CanalClientException
*/ */
void rollback(long batchId) throws CanalClientException; void rollback(long batchId) throws CanalClientException;
/** /**
* 回滚到未进行 {@link ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link ack} 的地方开始拿 * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
* *
* @throws CanalClientException * @throws CanalClientException
*/ */

1
common/pom.xml
View File

@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version> <version>1.0.20-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.common</artifactId> <artifactId>canal.common</artifactId>
<packaging>jar</packaging> <packaging>jar</packaging>
<name>canal common module for otter ${project.version}</name> <name>canal common module for otter ${project.version}</name>

6
common/src/main/java/com/alibaba/otter/canal/common/CanalLifeCycle.java
View File

@@ -6,9 +6,9 @@ package com.alibaba.otter.canal.common;
*/ */
public interface CanalLifeCycle { public interface CanalLifeCycle {
public void start(); void start();
public void stop(); void stop();
public boolean isStart(); boolean isStart();
} }

3
common/src/main/java/com/alibaba/otter/canal/common/alarm/CanalAlarmHandler.java
View File

@@ -14,8 +14,7 @@ public interface CanalAlarmHandler extends CanalLifeCycle {
* 发送对应destination的报警 * 发送对应destination的报警
* *
* @param destination * @param destination
* @param title
* @param msg * @param msg
*/ */
public void sendAlarm(String destination, String msg); void sendAlarm(String destination, String msg);
} }

13
common/src/main/java/com/alibaba/otter/canal/common/zookeeper/ZkClientx.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.common.zookeeper; package com.alibaba.otter.canal.common.zookeeper;
import com.google.common.collect.MigrateMap;
import java.util.Map; import java.util.Map;
import org.I0Itec.zkclient.IZkConnection; import org.I0Itec.zkclient.IZkConnection;
@@ -23,12 +24,14 @@ import com.google.common.collect.MapMaker;
public class ZkClientx extends ZkClient { public class ZkClientx extends ZkClient {
// 对于zkclient进行一次缓存，避免一个jvm内部使用多个zk connection // 对于zkclient进行一次缓存，避免一个jvm内部使用多个zk connection
private static Map<String, ZkClientx> clients = new MapMaker().makeComputingMap(new Function<String, ZkClientx>() { private static Map<String, ZkClientx> clients = MigrateMap.makeComputingMap(new Function<String, ZkClientx>()
{
public ZkClientx apply(String servers) { public ZkClientx apply(String servers)
return new ZkClientx(servers); {
} return new ZkClientx(servers);
}); }
});
public static ZkClientx getZkClient(String servers) { public static ZkClientx getZkClient(String servers) {
return clients.get(servers); return clients.get(servers);

17
common/src/main/java/com/google/common/collect/MigrateMap.java Normal file
View File

@@ -0,0 +1,17 @@
package com.google.common.collect;
import com.google.common.base.Function;
import java.util.concurrent.ConcurrentMap;
public class MigrateMap
{
public static <K, V> ConcurrentMap<K, V> makeComputingMap(MapMaker maker, Function<? super K, ? extends V> computingFunction)
{
return maker.makeComputingMap(computingFunction);
}
public static <K, V> ConcurrentMap<K, V> makeComputingMap(Function<? super K, ? extends V> computingFunction)
{
return new MapMaker().makeComputingMap(computingFunction);
}
}

104
deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.deployer; package com.alibaba.otter.canal.deployer;
import com.google.common.collect.MigrateMap;
import java.util.Map; import java.util.Map;
import java.util.Properties; import java.util.Properties;
@@ -33,7 +34,7 @@ import com.alibaba.otter.canal.instance.core.CanalInstanceGenerator;
import com.alibaba.otter.canal.instance.manager.CanalConfigClient; import com.alibaba.otter.canal.instance.manager.CanalConfigClient;
import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator; import com.alibaba.otter.canal.instance.manager.ManagerCanalInstanceGenerator;
import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator; import com.alibaba.otter.canal.instance.spring.SpringCanalInstanceGenerator;
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.exception.CanalServerException; import com.alibaba.otter.canal.server.exception.CanalServerException;
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty; import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
import com.google.common.base.Function; import com.google.common.base.Function;
@@ -59,7 +60,7 @@ public class CanalController {
private boolean autoScan = true; private boolean autoScan = true;
private InstanceAction defaultAction; private InstanceAction defaultAction;
private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors; private Map<InstanceMode, InstanceConfigMonitor> instanceConfigMonitors;
private CanalServerWithEmbeded embededCanalServer; private CanalServerWithEmbedded embededCanalServer;
private CanalServerWithNetty canalServer; private CanalServerWithNetty canalServer;
private CanalInstanceGenerator instanceGenerator; private CanalInstanceGenerator instanceGenerator;
@@ -70,9 +71,11 @@ public class CanalController {
} }
public CanalController(final Properties properties){ public CanalController(final Properties properties){
managerClients = new MapMaker().makeComputingMap(new Function<String, CanalConfigClient>() { managerClients = MigrateMap.makeComputingMap(new Function<String, CanalConfigClient>()
{
public CanalConfigClient apply(String managerAddress) { public CanalConfigClient apply(String managerAddress)
{
return getManagerClient(managerAddress); return getManagerClient(managerAddress);
} }
}); });
@@ -87,7 +90,7 @@ public class CanalController {
cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID)); cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
ip = getProperty(properties, CanalConstants.CANAL_IP); ip = getProperty(properties, CanalConstants.CANAL_IP);
port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT)); port = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_PORT));
embededCanalServer = new CanalServerWithEmbeded(); embededCanalServer = new CanalServerWithEmbedded();
embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator embededCanalServer.setCanalInstanceGenerator(instanceGenerator);// 设置自定义的instanceGenerator
canalServer = new CanalServerWithNetty(embededCanalServer); canalServer = new CanalServerWithNetty(embededCanalServer);
canalServer.setIp(ip); canalServer.setIp(ip);
@@ -107,68 +110,88 @@ public class CanalController {
final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port); final ServerRunningData serverData = new ServerRunningData(cid, ip + ":" + port);
ServerRunningMonitors.setServerData(serverData); ServerRunningMonitors.setServerData(serverData);
ServerRunningMonitors.setRunningMonitors(new MapMaker().makeComputingMap(new Function<String, ServerRunningMonitor>() { ServerRunningMonitors.setRunningMonitors(MigrateMap.makeComputingMap(new Function<String, ServerRunningMonitor>()
{
public ServerRunningMonitor apply(final String destination) { public ServerRunningMonitor apply(final String destination)
{
ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData); ServerRunningMonitor runningMonitor = new ServerRunningMonitor(serverData);
runningMonitor.setDestination(destination); runningMonitor.setDestination(destination);
runningMonitor.setListener(new ServerRunningListener() { runningMonitor.setListener(new ServerRunningListener()
{
public void processActiveEnter() { public void processActiveEnter()
try { {
try
{
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
embededCanalServer.start(destination); embededCanalServer.start(destination);
} finally { } finally
{
MDC.remove(CanalConstants.MDC_DESTINATION); MDC.remove(CanalConstants.MDC_DESTINATION);
} }
} }
public void processActiveExit() { public void processActiveExit()
try { {
try
{
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
embededCanalServer.stop(destination); embededCanalServer.stop(destination);
} finally { } finally
{
MDC.remove(CanalConstants.MDC_DESTINATION); MDC.remove(CanalConstants.MDC_DESTINATION);
} }
} }
public void processStart() { public void processStart()
try { {
if (zkclientx != null) { try
{
if (zkclientx != null)
{
final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":"
+ port); + port);
initCid(path); initCid(path);
zkclientx.subscribeStateChanges(new IZkStateListener() { zkclientx.subscribeStateChanges(new IZkStateListener()
{
public void handleStateChanged(KeeperState state) throws Exception { public void handleStateChanged(KeeperState state) throws Exception
{
} }
public void handleNewSession() throws Exception { public void handleNewSession() throws Exception
{
initCid(path); initCid(path);
} }
}); });
} }
} finally { } finally
{
MDC.remove(CanalConstants.MDC_DESTINATION); MDC.remove(CanalConstants.MDC_DESTINATION);
} }
} }
public void processStop() { public void processStop()
try { {
try
{
MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination)); MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
if (zkclientx != null) { if (zkclientx != null)
{
final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":" final String path = ZookeeperPathUtils.getDestinationClusterNode(destination, ip + ":"
+ port); + port);
releaseCid(path); releaseCid(path);
} }
} finally { } finally
{
MDC.remove(CanalConstants.MDC_DESTINATION); MDC.remove(CanalConstants.MDC_DESTINATION);
} }
} }
}); });
if (zkclientx != null) { if (zkclientx != null)
{
runningMonitor.setZkClient(zkclientx); runningMonitor.setZkClient(zkclientx);
} }
return runningMonitor; return runningMonitor;
@@ -216,25 +239,32 @@ public class CanalController {
} }
}; };
instanceConfigMonitors = new MapMaker().makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>() { instanceConfigMonitors = MigrateMap.makeComputingMap(new Function<InstanceMode, InstanceConfigMonitor>()
{
public InstanceConfigMonitor apply(InstanceMode mode) { public InstanceConfigMonitor apply(InstanceMode mode)
int scanInterval = Integer.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL)); {
int scanInterval = Integer
.valueOf(getProperty(properties, CanalConstants.CANAL_AUTO_SCAN_INTERVAL));
if (mode.isSpring()) { if (mode.isSpring())
{
SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor(); SpringInstanceConfigMonitor monitor = new SpringInstanceConfigMonitor();
monitor.setScanIntervalInSecond(scanInterval); monitor.setScanIntervalInSecond(scanInterval);
monitor.setDefaultAction(defaultAction); monitor.setDefaultAction(defaultAction);
// 设置conf目录，默认是user.dir + conf目录组成 // 设置conf目录，默认是user.dir + conf目录组成
String rootDir = getProperty(properties, CanalConstants.CANAL_CONF_DIR); String rootDir = getProperty(properties, CanalConstants.CANAL_CONF_DIR);
if (StringUtils.isEmpty(rootDir)) { if (StringUtils.isEmpty(rootDir))
{
rootDir = "../conf"; rootDir = "../conf";
} }
monitor.setRootConf(rootDir); monitor.setRootConf(rootDir);
return monitor; return monitor;
} else if (mode.isManager()) { } else if (mode.isManager())
{
return new ManagerInstanceConfigMonitor(); return new ManagerInstanceConfigMonitor();
} else { } else
{
throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor"); throw new UnsupportedOperationException("unknow mode :" + mode + " for monitor");
} }
} }
@@ -391,7 +421,7 @@ public class CanalController {
} }
if (autoScan) { if (autoScan) {
instanceConfigMonitors.get(config.getMode()).regeister(destination, defaultAction); instanceConfigMonitors.get(config.getMode()).register(destination, defaultAction);
} }
} }

6
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceAction.java
View File

@@ -11,15 +11,15 @@ public interface InstanceAction {
/** /**
* 启动destination * 启动destination
*/ */
public void start(String destination); void start(String destination);
/** /**
* 停止destination * 停止destination
*/ */
public void stop(String destination); void stop(String destination);
/** /**
* 重载destination，可能需要stop,start操作，或者只是更新下内存配置 * 重载destination，可能需要stop,start操作，或者只是更新下内存配置
*/ */
public void reload(String destination); void reload(String destination);
} }

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/InstanceConfigMonitor.java
View File

@@ -10,7 +10,7 @@ import com.alibaba.otter.canal.common.CanalLifeCycle;
*/ */
public interface InstanceConfigMonitor extends CanalLifeCycle { public interface InstanceConfigMonitor extends CanalLifeCycle {
public void regeister(String destination, InstanceAction action); void register(String destination, InstanceAction action);
public void unRegeister(String destination); void unregister(String destination);
} }

4
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/ManagerInstanceConfigMonitor.java
View File

@@ -9,11 +9,11 @@ import com.alibaba.otter.canal.common.CanalLifeCycle;
*/ */
public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, CanalLifeCycle { public class ManagerInstanceConfigMonitor extends AbstractCanalLifeCycle implements InstanceConfigMonitor, CanalLifeCycle {
public void regeister(String destination, InstanceAction action) { public void register(String destination, InstanceAction action) {
} }
public void unRegeister(String destination) { public void unregister(String destination) {
} }

17
deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.deployer.monitor; package com.alibaba.otter.canal.deployer.monitor;
import com.google.common.collect.MigrateMap;
import java.io.File; import java.io.File;
import java.io.FileFilter; import java.io.FileFilter;
import java.io.FilenameFilter; import java.io.FilenameFilter;
@@ -39,12 +40,14 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
private long scanIntervalInSecond = 5; private long scanIntervalInSecond = 5;
private InstanceAction defaultAction = null; private InstanceAction defaultAction = null;
private Map<String, InstanceAction> actions = new MapMaker().makeMap(); private Map<String, InstanceAction> actions = new MapMaker().makeMap();
private Map<String, InstanceConfigFiles> lastFiles = new MapMaker().makeComputingMap(new Function<String, InstanceConfigFiles>() { private Map<String, InstanceConfigFiles> lastFiles = MigrateMap.makeComputingMap(new Function<String, InstanceConfigFiles>()
{
public InstanceConfigFiles apply(String destination) { public InstanceConfigFiles apply(String destination)
return new InstanceConfigFiles(destination); {
} return new InstanceConfigFiles(destination);
}); }
});
private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1, private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1,
new NamedThreadFactory("canal-instance-scan")); new NamedThreadFactory("canal-instance-scan"));
@@ -72,7 +75,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
lastFiles.clear(); lastFiles.clear();
} }
public void regeister(String destination, InstanceAction action) { public void register(String destination, InstanceAction action) {
if (action != null) { if (action != null) {
actions.put(destination, action); actions.put(destination, action);
} else { } else {
@@ -80,7 +83,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen
} }
} }
public void unRegeister(String destination) { public void unregister(String destination) {
actions.remove(destination); actions.remove(destination);
} }

60
filter/src/main/java/com/alibaba/otter/canal/filter/PatternUtils.java
View File

@@ -1,15 +1,14 @@
package com.alibaba.otter.canal.filter; package com.alibaba.otter.canal.filter;
import java.util.Map;
import org.apache.oro.text.regex.MalformedPatternException;
import org.apache.oro.text.regex.Pattern;
import org.apache.oro.text.regex.PatternCompiler;
import org.apache.oro.text.regex.Perl5Compiler;
import com.alibaba.otter.canal.filter.exception.CanalFilterException; import com.alibaba.otter.canal.filter.exception.CanalFilterException;
import com.google.common.base.Function; import com.google.common.base.Function;
import com.google.common.collect.MapMaker; import com.google.common.collect.MapMaker;
import com.google.common.collect.MigrateMap;
import java.util.Map;
import org.apache.oro.text.regex.MalformedPatternException;
import org.apache.oro.text.regex.Pattern;
import org.apache.oro.text.regex.PatternCompiler;
import org.apache.oro.text.regex.Perl5Compiler;
/** /**
* 提供{@linkplain Pattern}的lazy get处理 * 提供{@linkplain Pattern}的lazy get处理
@@ -17,32 +16,39 @@ import com.google.common.collect.MapMaker;
* @author jianghang 2013-1-22 下午09:36:44 * @author jianghang 2013-1-22 下午09:36:44
* @version 1.0.0 * @version 1.0.0
*/ */
public class PatternUtils { public class PatternUtils
{
private static Map<String, Pattern> patterns = new MapMaker().softValues().makeComputingMap( private static Map<String, Pattern> patterns = MigrateMap.makeComputingMap(new MapMaker().softValues(),
new Function<String, Pattern>() { new Function<String, Pattern>()
{
public Pattern apply( public Pattern apply(
String pattern) { String pattern)
try { {
PatternCompiler pc = new Perl5Compiler(); try
return pc.compile( {
pattern, PatternCompiler pc = new Perl5Compiler();
Perl5Compiler.CASE_INSENSITIVE_MASK return pc.compile(
| Perl5Compiler.READ_ONLY_MASK pattern,
| Perl5Compiler.SINGLELINE_MASK); Perl5Compiler.CASE_INSENSITIVE_MASK
} catch (MalformedPatternException e) { | Perl5Compiler.READ_ONLY_MASK
throw new CanalFilterException( | Perl5Compiler.SINGLELINE_MASK);
e); } catch (MalformedPatternException e)
} {
} throw new CanalFilterException(
}); e);
}
}
});
public static Pattern getPattern(String pattern) { public static Pattern getPattern(String pattern)
{
return patterns.get(pattern); return patterns.get(pattern);
} }
public static void clear() { public static void clear()
{
patterns.clear(); patterns.clear();
} }
} }

1
instance/core/pom.xml
View File

@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version> <version>1.0.20-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath> <relativePath>../../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.instance.core</artifactId> <artifactId>canal.instance.core</artifactId>
<packaging>jar</packaging> <packaging>jar</packaging>
<name>canal instance core module for otter ${project.version}</name> <name>canal instance core module for otter ${project.version}</name>

14
instance/core/src/main/java/com/alibaba/otter/canal/instance/core/CanalInstance.java
View File

@@ -16,20 +16,20 @@ import com.alibaba.otter.canal.store.CanalEventStore;
*/ */
public interface CanalInstance extends CanalLifeCycle { public interface CanalInstance extends CanalLifeCycle {
public String getDestination(); String getDestination();
public CanalEventParser getEventParser(); CanalEventParser getEventParser();
public CanalEventSink getEventSink(); CanalEventSink getEventSink();
public CanalEventStore getEventStore(); CanalEventStore getEventStore();
public CanalMetaManager getMetaManager(); CanalMetaManager getMetaManager();
public CanalAlarmHandler getAlarmHandler(); CanalAlarmHandler getAlarmHandler();
/** /**
* 客户端发生订阅/取消订阅行为 * 客户端发生订阅/取消订阅行为
*/ */
public boolean subscribeChange(ClientIdentity identity); boolean subscribeChange(ClientIdentity identity);
} }

9
instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java
View File

@@ -89,8 +89,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
this.destination = canal.getName(); this.destination = canal.getName();
this.filter = filter; this.filter = filter;
logger.info("init CannalInstance for {}-{} with parameters:{}", logger.info("init CannalInstance for {}-{} with parameters:{}", canalId, destination, parameters);
new Object[] { canalId, destination, parameters });
// 初始化报警机制 // 初始化报警机制
initAlarmHandler(); initAlarmHandler();
// 初始化metaManager // 初始化metaManager
@@ -116,8 +115,8 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
public void start() { public void start() {
super.start(); super.start();
// 初始化metaManager // 初始化metaManager
logger.info("start CannalInstance for {}-{} with parameters:{}", new Object[] { canalId, destination, logger.info("start CannalInstance for {}-{} with parameters:{}", canalId, destination,
parameters }); parameters);
if (!metaManager.isStart()) { if (!metaManager.isStart()) {
metaManager.start(); metaManager.start();
@@ -336,7 +335,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca
} }
private CanalEventParser doInitEventParser(SourcingType type, List<InetSocketAddress> dbAddresses) { private CanalEventParser doInitEventParser(SourcingType type, List<InetSocketAddress> dbAddresses) {
CanalEventParser eventParser = null; CanalEventParser eventParser;
if (type.isMysql()) { if (type.isMysql()) {
MysqlEventParser mysqlEventParser = new MysqlEventParser(); MysqlEventParser mysqlEventParser = new MysqlEventParser();
mysqlEventParser.setDestination(destination); mysqlEventParser.setDestination(destination);

4
meta/src/main/java/com/alibaba/otter/canal/meta/CanalMetaManager.java
View File

@@ -35,12 +35,12 @@ public interface CanalMetaManager extends CanalLifeCycle {
void unsubscribe(ClientIdentity clientIdentity) throws CanalMetaManagerException; void unsubscribe(ClientIdentity clientIdentity) throws CanalMetaManagerException;
/** /**
* 获取cuosr游标 * 获取 cursor 游标
*/ */
Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException; Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException;
/** /**
* 更新cuosr游标 * 更新 cursor 游标
*/ */
void updateCursor(ClientIdentity clientIdentity, Position position) throws CanalMetaManagerException; void updateCursor(ClientIdentity clientIdentity, Position position) throws CanalMetaManagerException;

25
meta/src/main/java/com/alibaba/otter/canal/meta/FileMixedMetaManager.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.meta; package com.alibaba.otter.canal.meta;
import com.google.common.collect.MigrateMap;
import java.io.File; import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.nio.charset.Charset; import java.nio.charset.Charset;
@@ -69,28 +70,36 @@ public class FileMixedMetaManager extends MemoryMetaManager implements CanalMeta
throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write"); throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write");
} }
dataFileCaches = new MapMaker().makeComputingMap(new Function<String, File>() { dataFileCaches = MigrateMap.makeComputingMap(new Function<String, File>()
{
public File apply(String destination) { public File apply(String destination)
{
return getDataFile(destination); return getDataFile(destination);
} }
}); });
executor = Executors.newScheduledThreadPool(1); executor = Executors.newScheduledThreadPool(1);
destinations = new MapMaker().makeComputingMap(new Function<String, List<ClientIdentity>>() { destinations = MigrateMap.makeComputingMap(new Function<String, List<ClientIdentity>>()
{
public List<ClientIdentity> apply(String destination) { public List<ClientIdentity> apply(String destination)
{
return loadClientIdentity(destination); return loadClientIdentity(destination);
} }
}); });
cursors = new MapMaker().makeComputingMap(new Function<ClientIdentity, Position>() { cursors = MigrateMap.makeComputingMap(new Function<ClientIdentity, Position>()
{
public Position apply(ClientIdentity clientIdentity) { public Position apply(ClientIdentity clientIdentity)
{
Position position = loadCursor(clientIdentity.getDestination(), clientIdentity); Position position = loadCursor(clientIdentity.getDestination(), clientIdentity);
if (position == null) { if (position == null)
{
return nullCursor; // 返回一个空对象标识，避免出现异常 return nullCursor; // 返回一个空对象标识，避免出现异常
} else { } else
{
return position; return position;
} }
} }

13
meta/src/main/java/com/alibaba/otter/canal/meta/MemoryMetaManager.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.meta; package com.alibaba.otter.canal.meta;
import com.google.common.collect.MigrateMap;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@@ -31,9 +32,11 @@ public class MemoryMetaManager extends AbstractCanalLifeCycle implements CanalMe
public void start() { public void start() {
super.start(); super.start();
batches = new MapMaker().makeComputingMap(new Function<ClientIdentity, MemoryClientIdentityBatch>() { batches = MigrateMap.makeComputingMap(new Function<ClientIdentity, MemoryClientIdentityBatch>()
{
public MemoryClientIdentityBatch apply(ClientIdentity clientIdentity) { public MemoryClientIdentityBatch apply(ClientIdentity clientIdentity)
{
return MemoryClientIdentityBatch.create(clientIdentity); return MemoryClientIdentityBatch.create(clientIdentity);
} }
@@ -41,9 +44,11 @@ public class MemoryMetaManager extends AbstractCanalLifeCycle implements CanalMe
cursors = new MapMaker().makeMap(); cursors = new MapMaker().makeMap();
destinations = new MapMaker().makeComputingMap(new Function<String, List<ClientIdentity>>() { destinations = MigrateMap.makeComputingMap(new Function<String, List<ClientIdentity>>()
{
public List<ClientIdentity> apply(String destination) { public List<ClientIdentity> apply(String destination)
{
return Lists.newArrayList(); return Lists.newArrayList();
} }
}); });

138
meta/src/main/java/com/alibaba/otter/canal/meta/MixedMetaManager.java
View File

@@ -1,18 +1,16 @@
package com.alibaba.otter.canal.meta; package com.alibaba.otter.canal.meta;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.meta.exception.CanalMetaManagerException; import com.alibaba.otter.canal.meta.exception.CanalMetaManagerException;
import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.position.Position; import com.alibaba.otter.canal.protocol.position.Position;
import com.alibaba.otter.canal.protocol.position.PositionRange; import com.alibaba.otter.canal.protocol.position.PositionRange;
import com.google.common.base.Function; import com.google.common.base.Function;
import com.google.common.collect.MapMaker; import com.google.common.collect.MigrateMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import org.springframework.util.Assert;
/** /**
* 组合memory + zookeeper的使用模式 * 组合memory + zookeeper的使用模式
@@ -21,48 +19,61 @@ import com.google.common.collect.MapMaker;
* @version 1.0.0 * @version 1.0.0
*/ */
public class MixedMetaManager extends MemoryMetaManager implements CanalMetaManager { public class MixedMetaManager extends MemoryMetaManager implements CanalMetaManager
{
private ExecutorService executor; private ExecutorService executor;
private ZooKeeperMetaManager zooKeeperMetaManager; private ZooKeeperMetaManager zooKeeperMetaManager;
@SuppressWarnings("serial") @SuppressWarnings("serial")
private final Position nullCursor = new Position() { private final Position nullCursor = new Position()
}; {
};
public void start() { public void start()
{
super.start(); super.start();
Assert.notNull(zooKeeperMetaManager); Assert.notNull(zooKeeperMetaManager);
if (!zooKeeperMetaManager.isStart()) { if (!zooKeeperMetaManager.isStart())
{
zooKeeperMetaManager.start(); zooKeeperMetaManager.start();
} }
executor = Executors.newFixedThreadPool(1); executor = Executors.newFixedThreadPool(1);
destinations = new MapMaker().makeComputingMap(new Function<String, List<ClientIdentity>>() { destinations = MigrateMap.makeComputingMap(new Function<String, List<ClientIdentity>>()
{
public List<ClientIdentity> apply(String destination) { public List<ClientIdentity> apply(String destination)
{
return zooKeeperMetaManager.listAllSubscribeInfo(destination); return zooKeeperMetaManager.listAllSubscribeInfo(destination);
} }
}); });
cursors = new MapMaker().makeComputingMap(new Function<ClientIdentity, Position>() { cursors = MigrateMap.makeComputingMap(new Function<ClientIdentity, Position>()
{
public Position apply(ClientIdentity clientIdentity) { public Position apply(ClientIdentity clientIdentity)
{
Position position = zooKeeperMetaManager.getCursor(clientIdentity); Position position = zooKeeperMetaManager.getCursor(clientIdentity);
if (position == null) { if (position == null)
{
return nullCursor; // 返回一个空对象标识，避免出现异常 return nullCursor; // 返回一个空对象标识，避免出现异常
} else { } else
{
return position; return position;
} }
} }
}); });
batches = new MapMaker().makeComputingMap(new Function<ClientIdentity, MemoryClientIdentityBatch>() { batches = MigrateMap.makeComputingMap(new Function<ClientIdentity, MemoryClientIdentityBatch>()
{
public MemoryClientIdentityBatch apply(ClientIdentity clientIdentity) { public MemoryClientIdentityBatch apply(ClientIdentity clientIdentity)
{
// 读取一下zookeeper信息，初始化一次 // 读取一下zookeeper信息，初始化一次
MemoryClientIdentityBatch batches = MemoryClientIdentityBatch.create(clientIdentity); MemoryClientIdentityBatch batches = MemoryClientIdentityBatch.create(clientIdentity);
Map<Long, PositionRange> positionRanges = zooKeeperMetaManager.listAllBatchs(clientIdentity); Map<Long, PositionRange> positionRanges = zooKeeperMetaManager.listAllBatchs(clientIdentity);
for (Map.Entry<Long, PositionRange> entry : positionRanges.entrySet()) { for (Map.Entry<Long, PositionRange> entry : positionRanges.entrySet())
{
batches.addPositionRange(entry.getValue(), entry.getKey()); // 添加记录到指定batchId batches.addPositionRange(entry.getValue(), entry.getKey()); // 添加记录到指定batchId
} }
return batches; return batches;
@@ -70,10 +81,12 @@ public class MixedMetaManager extends MemoryMetaManager implements CanalMetaMana
}); });
} }
public void stop() { public void stop()
{
super.stop(); super.stop();
if (zooKeeperMetaManager.isStart()) { if (zooKeeperMetaManager.isStart())
{
zooKeeperMetaManager.stop(); zooKeeperMetaManager.stop();
} }
@@ -82,58 +95,73 @@ public class MixedMetaManager extends MemoryMetaManager implements CanalMetaMana
batches.clear(); batches.clear();
} }
public void subscribe(final ClientIdentity clientIdentity) throws CanalMetaManagerException { public void subscribe(final ClientIdentity clientIdentity) throws CanalMetaManagerException
{
super.subscribe(clientIdentity); super.subscribe(clientIdentity);
executor.submit(new Runnable() { executor.submit(new Runnable()
{
public void run() { public void run()
{
zooKeeperMetaManager.subscribe(clientIdentity); zooKeeperMetaManager.subscribe(clientIdentity);
} }
}); });
} }
public void unsubscribe(final ClientIdentity clientIdentity) throws CanalMetaManagerException { public void unsubscribe(final ClientIdentity clientIdentity) throws CanalMetaManagerException
{
super.unsubscribe(clientIdentity); super.unsubscribe(clientIdentity);
executor.submit(new Runnable() { executor.submit(new Runnable()
{
public void run() { public void run()
{
zooKeeperMetaManager.unsubscribe(clientIdentity); zooKeeperMetaManager.unsubscribe(clientIdentity);
} }
}); });
} }
public void updateCursor(final ClientIdentity clientIdentity, final Position position) public void updateCursor(final ClientIdentity clientIdentity, final Position position)
throws CanalMetaManagerException { throws CanalMetaManagerException
{
super.updateCursor(clientIdentity, position); super.updateCursor(clientIdentity, position);
// 异步刷新 // 异步刷新
executor.submit(new Runnable() { executor.submit(new Runnable()
{
public void run() { public void run()
{
zooKeeperMetaManager.updateCursor(clientIdentity, position); zooKeeperMetaManager.updateCursor(clientIdentity, position);
} }
}); });
} }
@Override @Override
public Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException { public Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException
{
Position position = super.getCursor(clientIdentity); Position position = super.getCursor(clientIdentity);
if (position == nullCursor) { if (position == nullCursor)
{
return null; return null;
} else { } else
{
return position; return position;
} }
} }
public Long addBatch(final ClientIdentity clientIdentity, final PositionRange positionRange) public Long addBatch(final ClientIdentity clientIdentity, final PositionRange positionRange)
throws CanalMetaManagerException { throws CanalMetaManagerException
{
final Long batchId = super.addBatch(clientIdentity, positionRange); final Long batchId = super.addBatch(clientIdentity, positionRange);
// 异步刷新 // 异步刷新
executor.submit(new Runnable() { executor.submit(new Runnable()
{
public void run() { public void run()
{
zooKeeperMetaManager.addBatch(clientIdentity, positionRange, batchId); zooKeeperMetaManager.addBatch(clientIdentity, positionRange, batchId);
} }
}); });
@@ -141,24 +169,30 @@ public class MixedMetaManager extends MemoryMetaManager implements CanalMetaMana
} }
public void addBatch(final ClientIdentity clientIdentity, final PositionRange positionRange, final Long batchId) public void addBatch(final ClientIdentity clientIdentity, final PositionRange positionRange, final Long batchId)
throws CanalMetaManagerException { throws CanalMetaManagerException
{
super.addBatch(clientIdentity, positionRange, batchId); super.addBatch(clientIdentity, positionRange, batchId);
// 异步刷新 // 异步刷新
executor.submit(new Runnable() { executor.submit(new Runnable()
{
public void run() { public void run()
{
zooKeeperMetaManager.addBatch(clientIdentity, positionRange, batchId); zooKeeperMetaManager.addBatch(clientIdentity, positionRange, batchId);
} }
}); });
} }
public PositionRange removeBatch(final ClientIdentity clientIdentity, final Long batchId) public PositionRange removeBatch(final ClientIdentity clientIdentity, final Long batchId)
throws CanalMetaManagerException { throws CanalMetaManagerException
{
PositionRange positionRange = super.removeBatch(clientIdentity, batchId); PositionRange positionRange = super.removeBatch(clientIdentity, batchId);
// 异步刷新 // 异步刷新
executor.submit(new Runnable() { executor.submit(new Runnable()
{
public void run() { public void run()
{
zooKeeperMetaManager.removeBatch(clientIdentity, batchId); zooKeeperMetaManager.removeBatch(clientIdentity, batchId);
} }
}); });
@@ -166,20 +200,24 @@ public class MixedMetaManager extends MemoryMetaManager implements CanalMetaMana
return positionRange; return positionRange;
} }
public void clearAllBatchs(final ClientIdentity clientIdentity) throws CanalMetaManagerException { public void clearAllBatchs(final ClientIdentity clientIdentity) throws CanalMetaManagerException
{
super.clearAllBatchs(clientIdentity); super.clearAllBatchs(clientIdentity);
// 异步刷新 // 异步刷新
executor.submit(new Runnable() { executor.submit(new Runnable()
{
public void run() { public void run()
{
zooKeeperMetaManager.clearAllBatchs(clientIdentity); zooKeeperMetaManager.clearAllBatchs(clientIdentity);
} }
}); });
} }
// =============== setter / getter ================ // =============== setter / getter ================
public void setZooKeeperMetaManager(ZooKeeperMetaManager zooKeeperMetaManager) { public void setZooKeeperMetaManager(ZooKeeperMetaManager zooKeeperMetaManager)
{
this.zooKeeperMetaManager = zooKeeperMetaManager; this.zooKeeperMetaManager = zooKeeperMetaManager;
} }
} }

29
meta/src/main/java/com/alibaba/otter/canal/meta/PeriodMixedMetaManager.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.meta; package com.alibaba.otter.canal.meta;
import com.google.common.collect.MigrateMap;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
@@ -52,32 +53,40 @@ public class PeriodMixedMetaManager extends MemoryMetaManager implements CanalMe
} }
executor = Executors.newScheduledThreadPool(1); executor = Executors.newScheduledThreadPool(1);
destinations = new MapMaker().makeComputingMap(new Function<String, List<ClientIdentity>>() { destinations = MigrateMap.makeComputingMap(new Function<String, List<ClientIdentity>>()
{
public List<ClientIdentity> apply(String destination) { public List<ClientIdentity> apply(String destination)
{
return zooKeeperMetaManager.listAllSubscribeInfo(destination); return zooKeeperMetaManager.listAllSubscribeInfo(destination);
} }
}); });
cursors = new MapMaker().makeComputingMap(new Function<ClientIdentity, Position>() { cursors = MigrateMap.makeComputingMap(new Function<ClientIdentity, Position>()
{
public Position apply(ClientIdentity clientIdentity) { public Position apply(ClientIdentity clientIdentity)
{
Position position = zooKeeperMetaManager.getCursor(clientIdentity); Position position = zooKeeperMetaManager.getCursor(clientIdentity);
if (position == null) { if (position == null)
{
return nullCursor; // 返回一个空对象标识，避免出现异常 return nullCursor; // 返回一个空对象标识，避免出现异常
} else { } else
{
return position; return position;
} }
} }
}); });
batches = new MapMaker().makeComputingMap(new Function<ClientIdentity, MemoryClientIdentityBatch>() { batches = MigrateMap.makeComputingMap(new Function<ClientIdentity, MemoryClientIdentityBatch>()
{
public MemoryClientIdentityBatch apply(ClientIdentity clientIdentity) { public MemoryClientIdentityBatch apply(ClientIdentity clientIdentity)
{
// 读取一下zookeeper信息，初始化一次 // 读取一下zookeeper信息，初始化一次
MemoryClientIdentityBatch batches = MemoryClientIdentityBatch.create(clientIdentity); MemoryClientIdentityBatch batches = MemoryClientIdentityBatch.create(clientIdentity);
Map<Long, PositionRange> positionRanges = zooKeeperMetaManager.listAllBatchs(clientIdentity); Map<Long, PositionRange> positionRanges = zooKeeperMetaManager.listAllBatchs(clientIdentity);
for (Map.Entry<Long, PositionRange> entry : positionRanges.entrySet()) { for (Map.Entry<Long, PositionRange> entry : positionRanges.entrySet())
{
batches.addPositionRange(entry.getValue(), entry.getKey()); // 添加记录到指定batchId batches.addPositionRange(entry.getValue(), entry.getKey()); // 添加记录到指定batchId
} }
return batches; return batches;

1
parse/pom.xml
View File

@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version> <version>1.0.20-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.parse</artifactId> <artifactId>canal.parse</artifactId>
<packaging>jar</packaging> <packaging>jar</packaging>
<name>canal parse module for otter ${project.version}</name> <name>canal parse module for otter ${project.version}</name>

19
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.parse.inbound.mysql.dbsync; package com.alibaba.otter.canal.parse.inbound.mysql.dbsync;
import com.google.common.collect.MigrateMap;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
@@ -38,17 +39,23 @@ public class TableMetaCache {
public TableMetaCache(MysqlConnection con){ public TableMetaCache(MysqlConnection con){
this.connection = con; this.connection = con;
tableMetaCache = new MapMaker().makeComputingMap(new Function<String, TableMeta>() { tableMetaCache = MigrateMap.makeComputingMap(new Function<String, TableMeta>()
{
public TableMeta apply(String name) { public TableMeta apply(String name)
try { {
try
{
return getTableMeta0(name); return getTableMeta0(name);
} catch (IOException e) { } catch (IOException e)
{
// 尝试做一次retry操作 // 尝试做一次retry操作
try { try
{
connection.reconnect(); connection.reconnect();
return getTableMeta0(name); return getTableMeta0(name);
} catch (IOException e1) { } catch (IOException e1)
{
throw new CanalParseException("fetch failed by table meta:" + name, e1); throw new CanalParseException("fetch failed by table meta:" + name, e1);
} }
} }

19
parse/src/main/java/com/alibaba/otter/canal/parse/index/FileMixedLogPositionManager.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.parse.index; package com.alibaba.otter.canal.parse.index;
import com.google.common.collect.MigrateMap;
import java.io.File; import java.io.File;
import java.io.IOException; import java.io.IOException;
import java.nio.charset.Charset; import java.nio.charset.Charset;
@@ -67,21 +68,27 @@ public class FileMixedLogPositionManager extends MemoryLogPositionManager {
throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write"); throw new CanalMetaManagerException("dir[" + dataDir.getPath() + "] can not read/write");
} }
dataFileCaches = new MapMaker().makeComputingMap(new Function<String, File>() { dataFileCaches = MigrateMap.makeComputingMap(new Function<String, File>()
{
public File apply(String destination) { public File apply(String destination)
{
return getDataFile(destination); return getDataFile(destination);
} }
}); });
executor = Executors.newScheduledThreadPool(1); executor = Executors.newScheduledThreadPool(1);
positions = new MapMaker().makeComputingMap(new Function<String, LogPosition>() { positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>()
{
public LogPosition apply(String destination) { public LogPosition apply(String destination)
{
LogPosition logPosition = loadDataFromFile(dataFileCaches.get(destination)); LogPosition logPosition = loadDataFromFile(dataFileCaches.get(destination));
if (logPosition == null) { if (logPosition == null)
{
return nullPosition; return nullPosition;
} else { } else
{
return logPosition; return logPosition;
} }
} }

13
parse/src/main/java/com/alibaba/otter/canal/parse/index/MixedLogPositionManager.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.parse.index; package com.alibaba.otter.canal.parse.index;
import com.google.common.collect.MigrateMap;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
@@ -34,13 +35,17 @@ public class MixedLogPositionManager extends MemoryLogPositionManager implements
zooKeeperLogPositionManager.start(); zooKeeperLogPositionManager.start();
} }
executor = Executors.newFixedThreadPool(1); executor = Executors.newFixedThreadPool(1);
positions = new MapMaker().makeComputingMap(new Function<String, LogPosition>() { positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>()
{
public LogPosition apply(String destination) { public LogPosition apply(String destination)
{
LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination); LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
if (logPosition == null) { if (logPosition == null)
{
return nullPosition; return nullPosition;
} else { } else
{
return logPosition; return logPosition;
} }
} }

13
parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.parse.index; package com.alibaba.otter.canal.parse.index;
import com.google.common.collect.MigrateMap;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.HashSet; import java.util.HashSet;
@@ -43,13 +44,17 @@ public class PeriodMixedLogPositionManager extends MemoryLogPositionManager impl
zooKeeperLogPositionManager.start(); zooKeeperLogPositionManager.start();
} }
executor = Executors.newScheduledThreadPool(1); executor = Executors.newScheduledThreadPool(1);
positions = new MapMaker().makeComputingMap(new Function<String, LogPosition>() { positions = MigrateMap.makeComputingMap(new Function<String, LogPosition>()
{
public LogPosition apply(String destination) { public LogPosition apply(String destination)
{
LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination); LogPosition logPosition = zooKeeperLogPositionManager.getLatestIndexBy(destination);
if (logPosition == null) { if (logPosition == null)
{
return nullPosition; return nullPosition;
} else { } else
{
return logPosition; return logPosition;
} }
} }

665
pom.xml
View File

@@ -1,311 +1,312 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
<modelVersion>4.0.0</modelVersion> xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<groupId>com.alibaba.otter</groupId> <modelVersion>4.0.0</modelVersion>
<artifactId>canal</artifactId> <groupId>com.alibaba.otter</groupId>
<packaging>pom</packaging> <artifactId>canal</artifactId>
<name>canal module for otter ${project.version}</name> <packaging>pom</packaging>
<version>1.0.20-SNAPSHOT</version> <name>canal module for otter ${project.version}</name>
<url>https://github.com/alibaba/canal</url> <version>1.0.20-SNAPSHOT</version>
<parent> <url>https://github.com/alibaba/canal</url>
<groupId>org.sonatype.oss</groupId> <parent>
<artifactId>oss-parent</artifactId> <groupId>org.sonatype.oss</groupId>
<version>7</version> <artifactId>oss-parent</artifactId>
</parent> <version>7</version>
<developers> </parent>
<developer> <developers>
<name>agapple</name> <developer>
<url>http://agapple.iteye.com</url> <name>agapple</name>
<email>jianghang115@gmail.com</email> <url>http://agapple.iteye.com</url>
<timezone>8</timezone> <email>jianghang115@gmail.com</email>
</developer> <timezone>8</timezone>
<developer> </developer>
<name>zavakid</name> <developer>
<url>http://www.zavakid.com</url> <name>zavakid</name>
<email>zava.kid@gmail.com</email> <url>http://www.zavakid.com</url>
<timezone>8</timezone> <email>zava.kid@gmail.com</email>
</developer> <timezone>8</timezone>
<developer> </developer>
<name>in355hz</name> <developer>
<url>http://in355hz.iteye.com</url> <name>in355hz</name>
<email>in355hz@gmail.com</email> <url>http://in355hz.iteye.com</url>
<timezone>8</timezone> <email>in355hz@gmail.com</email>
</developer> <timezone>8</timezone>
</developers> </developer>
</developers>
<licenses> <licenses>
<license> <license>
<name>Apache License, Version 2.0</name> <name>Apache License, Version 2.0</name>
<url>http://www.apache.org/licenses/LICENSE-2.0</url> <url>http://www.apache.org/licenses/LICENSE-2.0</url>
</license> </license>
</licenses> </licenses>
<scm> <scm>
<url>git@github.com:alibaba/canal.git</url> <url>git@github.com:alibaba/canal.git</url>
<connection>scm:git:git@github.com:alibaba/canal.git</connection> <connection>scm:git:git@github.com:alibaba/canal.git</connection>
<developerConnection>scm:git:git@github.com:alibaba/canal.git</developerConnection> <developerConnection>scm:git:git@github.com:alibaba/canal.git</developerConnection>
</scm> </scm>
<repositories> <repositories>
<repository> <repository>
<id>central</id> <id>central</id>
<url>http://repo1.maven.org/maven2</url> <url>http://repo1.maven.org/maven2</url>
<releases> <releases>
<enabled>true</enabled> <enabled>true</enabled>
</releases> </releases>
<snapshots> <snapshots>
<enabled>false</enabled> <enabled>false</enabled>
</snapshots> </snapshots>
</repository> </repository>
<repository> <repository>
<id>java.net</id> <id>java.net</id>
<url>http://download.java.net/maven/2/</url> <url>http://download.java.net/maven/2/</url>
<releases> <releases>
<enabled>true</enabled> <enabled>true</enabled>
</releases> </releases>
<snapshots> <snapshots>
<enabled>false</enabled> <enabled>false</enabled>
</snapshots> </snapshots>
</repository> </repository>
<repository> <repository>
<id>alibaba</id> <id>alibaba</id>
<url>http://code.alibabatech.com/mvn/releases/</url> <url>http://code.alibabatech.com/mvn/releases/</url>
<releases> <releases>
<enabled>true</enabled> <enabled>true</enabled>
</releases> </releases>
<snapshots> <snapshots>
<enabled>false</enabled> <enabled>false</enabled>
</snapshots> </snapshots>
</repository> </repository>
<repository> <repository>
<id>sonatype</id> <id>sonatype</id>
<name>sonatype</name> <name>sonatype</name>
<url>https://oss.sonatype.org/content/repositories/snapshots</url> <url>https://oss.sonatype.org/content/repositories/snapshots</url>
<releases> <releases>
<enabled>false</enabled> <enabled>false</enabled>
</releases> </releases>
<snapshots> <snapshots>
<enabled>true</enabled> <enabled>true</enabled>
</snapshots> </snapshots>
</repository> </repository>
<repository> <repository>
<id>sonatype-release</id> <id>sonatype-release</id>
<name>sonatype-release</name> <name>sonatype-release</name>
<url>https://oss.sonatype.org/service/local/repositories/releases/content</url> <url>https://oss.sonatype.org/service/local/repositories/releases/content</url>
<releases> <releases>
<enabled>false</enabled> <enabled>false</enabled>
</releases> </releases>
<snapshots> <snapshots>
<enabled>true</enabled> <enabled>true</enabled>
</snapshots> </snapshots>
</repository> </repository>
</repositories> </repositories>
<properties> <properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<!--maven properties--> <!--maven properties-->
<maven.test.skip>true</maven.test.skip> <maven.test.skip>true</maven.test.skip>
<downloadSources>true</downloadSources> <downloadSources>true</downloadSources>
<!-- compiler settings properties --> <!-- compiler settings properties -->
<java_source_version>1.6</java_source_version> <java_source_version>1.6</java_source_version>
<java_target_version>1.6</java_target_version> <java_target_version>1.6</java_target_version>
<file_encoding>UTF-8</file_encoding> <file_encoding>UTF-8</file_encoding>
</properties> </properties>
<modules> <modules>
<module>common</module> <module>common</module>
<module>meta</module> <module>meta</module>
<module>dbsync</module> <module>dbsync</module>
<module>filter</module> <module>filter</module>
<module>driver</module> <module>driver</module>
<module>parse</module> <module>parse</module>
<module>sink</module> <module>sink</module>
<module>store</module> <module>store</module>
<module>protocol</module> <module>protocol</module>
<module>instance</module> <module>instance</module>
<module>server</module> <module>server</module>
<module>client</module> <module>client</module>
<module>deployer</module> <module>deployer</module>
<module>example</module> <module>example</module>
</modules> </modules>
<dependencyManagement> <dependencyManagement>
<dependencies> <dependencies>
<dependency> <dependency>
<groupId>org.springframework</groupId> <groupId>org.springframework</groupId>
<artifactId>spring</artifactId> <artifactId>spring</artifactId>
<version>2.5.6</version> <version>2.5.6</version>
</dependency> </dependency>
<!-- external --> <!-- external -->
<dependency> <dependency>
<groupId>commons-lang</groupId> <groupId>commons-lang</groupId>
<artifactId>commons-lang</artifactId> <artifactId>commons-lang</artifactId>
<version>2.6</version> <version>2.6</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>commons-io</groupId> <groupId>commons-io</groupId>
<artifactId>commons-io</artifactId> <artifactId>commons-io</artifactId>
<version>2.4</version> <version>2.4</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.apache.zookeeper</groupId> <groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId> <artifactId>zookeeper</artifactId>
<version>3.4.5</version> <version>3.4.5</version>
<exclusions> <exclusions>
<exclusion> <exclusion>
<groupId>log4j</groupId> <groupId>log4j</groupId>
<artifactId>log4j</artifactId> <artifactId>log4j</artifactId>
</exclusion> </exclusion>
<exclusion> <exclusion>
<groupId>org.slf4j</groupId> <groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId> <artifactId>slf4j-log4j12</artifactId>
</exclusion> </exclusion>
<exclusion> <exclusion>
<groupId>org.slf4j</groupId> <groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId> <artifactId>slf4j-api</artifactId>
</exclusion> </exclusion>
<exclusion> <exclusion>
<groupId>jline</groupId> <groupId>jline</groupId>
<artifactId>jline</artifactId> <artifactId>jline</artifactId>
</exclusion> </exclusion>
</exclusions> </exclusions>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.github.sgroschupf</groupId> <groupId>com.github.sgroschupf</groupId>
<artifactId>zkclient</artifactId> <artifactId>zkclient</artifactId>
<version>0.1</version> <version>0.1</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.alibaba</groupId> <groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId> <artifactId>fastjson</artifactId>
<version>1.1.26</version> <version>1.1.26</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.google.guava</groupId> <groupId>com.google.guava</groupId>
<artifactId>guava</artifactId> <artifactId>guava</artifactId>
<version>r09</version> <version>18.0</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.googlecode.aviator</groupId> <groupId>com.googlecode.aviator</groupId>
<artifactId>aviator</artifactId> <artifactId>aviator</artifactId>
<version>2.2.1</version> <version>2.2.1</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>oro</groupId> <groupId>oro</groupId>
<artifactId>oro</artifactId> <artifactId>oro</artifactId>
<version>2.0.8</version> <version>2.0.8</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.jboss.netty</groupId> <groupId>org.jboss.netty</groupId>
<artifactId>netty</artifactId> <artifactId>netty</artifactId>
<version>3.2.5.Final</version> <version>3.2.5.Final</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>com.google.protobuf</groupId> <groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId> <artifactId>protobuf-java</artifactId>
<version>2.4.1</version> <version>2.4.1</version>
</dependency> </dependency>
<!-- log --> <!-- log -->
<dependency> <dependency>
<groupId>ch.qos.logback</groupId> <groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId> <artifactId>logback-core</artifactId>
<version>1.0.6</version> <version>1.1.3</version>
</dependency> </dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.0.6</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jcl-over-slf4j</artifactId>
<version>1.6.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.6.0</version>
</dependency>
<!-- test dependency -->
<dependency> <dependency>
<groupId>junit</groupId> <groupId>ch.qos.logback</groupId>
<artifactId>junit</artifactId> <artifactId>logback-classic</artifactId>
<version>4.5</version> <version>1.1.3</version>
<scope>test</scope> </dependency>
</dependency> <dependency>
<dependency> <groupId>org.slf4j</groupId>
<groupId>mysql</groupId> <artifactId>jcl-over-slf4j</artifactId>
<artifactId>mysql-connector-java</artifactId> <version>1.7.12</version>
<version>5.1.12</version> </dependency>
<scope>test</scope> <dependency>
</dependency> <groupId>org.slf4j</groupId>
</dependencies> <artifactId>slf4j-api</artifactId>
</dependencyManagement> <version>1.7.12</version>
</dependency>
<!-- test dependency -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.12</version>
<scope>test</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build> <build>
<extensions> <extensions>
<extension> <extension>
<groupId>org.jvnet.wagon-svn</groupId> <groupId>org.jvnet.wagon-svn</groupId>
<artifactId>wagon-svn</artifactId> <artifactId>wagon-svn</artifactId>
<version>1.9</version> <version>1.9</version>
</extension> </extension>
<extension> <extension>
<groupId>org.apache.maven.wagon</groupId> <groupId>org.apache.maven.wagon</groupId>
<artifactId>wagon-http-shared</artifactId> <artifactId>wagon-http-shared</artifactId>
<version>1.0-beta-7</version> <version>1.0-beta-7</version>
</extension> </extension>
</extensions> </extensions>
<plugins> <plugins>
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-source-plugin</artifactId> <artifactId>maven-source-plugin</artifactId>
<version>2.4</version> <version>2.4</version>
<executions> <executions>
<execution> <execution>
<id>attach-sources</id> <id>attach-sources</id>
<goals> <goals>
<goal>jar</goal> <goal>jar</goal>
</goals> </goals>
</execution> </execution>
</executions> </executions>
</plugin> </plugin>
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId> <artifactId>maven-compiler-plugin</artifactId>
<version>3.2</version> <version>3.2</version>
<configuration> <configuration>
<source>${java_source_version}</source> <source>${java_source_version}</source>
<target>${java_target_version}</target> <target>${java_target_version}</target>
<encoding>${file_encoding}</encoding> <encoding>${file_encoding}</encoding>
</configuration> </configuration>
</plugin> </plugin>
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-eclipse-plugin</artifactId> <artifactId>maven-eclipse-plugin</artifactId>
<version>2.5.1</version> <version>2.5.1</version>
<configuration> <configuration>
<additionalConfig> <additionalConfig>
<file> <file>
<name>.settings/org.eclipse.core.resources.prefs</name> <name>.settings/org.eclipse.core.resources.prefs</name>
<content> <content>
<![CDATA[eclipse.preferences.version=1${line.separator}encoding/<project>=${file_encoding}${line.separator}]]> <![CDATA[eclipse.preferences.version=1${line.separator}encoding/<project>=${file_encoding}${line.separator}]]>
</content> </content>
</file> </file>
</additionalConfig> </additionalConfig>
</configuration> </configuration>
</plugin> </plugin>
<plugin> <plugin>
<groupId>org.apache.maven.plugins</groupId> <groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId> <artifactId>maven-surefire-plugin</artifactId>
<version>2.5</version> <version>2.5</version>
<configuration> <configuration>
<includes> <includes>
<include>**/*Test.java</include> <include>**/*Test.java</include>
</includes> </includes>
<excludes> <excludes>
<exclude>**/*NoRunTest.java</exclude> <exclude>**/*NoRunTest.java</exclude>
</excludes> </excludes>
</configuration> </configuration>
</plugin> </plugin>
<!-- javadoc --> <!-- javadoc -->
<!-- <!--
<plugin> <plugin>
@@ -362,43 +363,43 @@
</executions> </executions>
</plugin> </plugin>
--> -->
</plugins> </plugins>
<sourceDirectory>src/main/java</sourceDirectory> <sourceDirectory>src/main/java</sourceDirectory>
<testSourceDirectory>src/test/java</testSourceDirectory> <testSourceDirectory>src/test/java</testSourceDirectory>
<resources> <resources>
<resource> <resource>
<directory>src/main/resources</directory> <directory>src/main/resources</directory>
<includes> <includes>
<include>**/*</include> <include>**/*</include>
</includes> </includes>
<excludes> <excludes>
<exclude>**/.svn/</exclude> <exclude>**/.svn/</exclude>
</excludes> </excludes>
</resource> </resource>
</resources> </resources>
<testResources> <testResources>
<testResource> <testResource>
<directory>src/test/resources</directory> <directory>src/test/resources</directory>
<includes> <includes>
<include>**/*</include> <include>**/*</include>
</includes> </includes>
<excludes> <excludes>
<exclude>**/.svn/</exclude> <exclude>**/.svn/</exclude>
</excludes> </excludes>
</testResource> </testResource>
</testResources> </testResources>
</build> </build>
<distributionManagement> <distributionManagement>
<snapshotRepository> <snapshotRepository>
<id>sonatype-nexus-snapshots</id> <id>sonatype-nexus-snapshots</id>
<name>Sonatype Nexus Snapshots</name> <name>Sonatype Nexus Snapshots</name>
<url>https://oss.sonatype.org/content/repositories/snapshots/</url> <url>https://oss.sonatype.org/content/repositories/snapshots/</url>
</snapshotRepository> </snapshotRepository>
<repository> <repository>
<id>sonatype-nexus-staging</id> <id>sonatype-nexus-staging</id>
<name>Nexus Release Repository</name> <name>Nexus Release Repository</name>
<url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url> <url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url>
</repository> </repository>
</distributionManagement> </distributionManagement>
</project> </project>

4
protocol/src/main/java/com/alibaba/otter/canal/protocol/ClientIdentity.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.protocol; package com.alibaba.otter.canal.protocol;
import java.io.Serializable;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.builder.ToStringBuilder; import org.apache.commons.lang.builder.ToStringBuilder;
@@ -9,7 +10,8 @@ import com.alibaba.otter.canal.common.utils.CanalToStringStyle;
* @author zebin.xuzb @ 2012-6-20 * @author zebin.xuzb @ 2012-6-20
* @version 1.0.0 * @version 1.0.0
*/ */
public class ClientIdentity { public class ClientIdentity implements Serializable
{
private String destination; private String destination;
private short clientId; private short clientId;

1
server/pom.xml
View File

@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version> <version>1.0.20-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath> <relativePath>../pom.xml</relativePath>
</parent> </parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.server</artifactId> <artifactId>canal.server</artifactId>
<packaging>jar</packaging> <packaging>jar</packaging>
<name>canal server module for otter ${project.version}</name> <name>canal server module for otter ${project.version}</name>

4
server/src/main/java/com/alibaba/otter/canal/server/CanalServer.java
View File

@@ -11,7 +11,7 @@ import com.alibaba.otter.canal.server.exception.CanalServerException;
*/ */
public interface CanalServer extends CanalLifeCycle { public interface CanalServer extends CanalLifeCycle {
public void start() throws CanalServerException; void start() throws CanalServerException;
public void stop() throws CanalServerException; void stop() throws CanalServerException;
} }

29
server/src/main/java/com/alibaba/otter/canal/server/CanalService.java Normal file
View File

@@ -0,0 +1,29 @@
package com.alibaba.otter.canal.server;
import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.exception.CanalServerException;
import java.util.concurrent.TimeUnit;
public interface CanalService
{
void subscribe(ClientIdentity clientIdentity) throws CanalServerException;
void unsubscribe(ClientIdentity clientIdentity) throws CanalServerException;
Message get(ClientIdentity clientIdentity, int batchSize) throws CanalServerException;
Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
throws CanalServerException;
Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException;
Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
throws CanalServerException;
void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException;
void rollback(ClientIdentity clientIdentity) throws CanalServerException;
void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException;
}

45
server/src/main/java/com/alibaba/otter/canal/server/embeded/CanalServerWithEmbeded.java → server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.server.embeded; package com.alibaba.otter.canal.server.embedded;
import com.google.common.collect.MigrateMap;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
@@ -27,7 +28,6 @@ import com.alibaba.otter.canal.store.model.Event;
import com.alibaba.otter.canal.store.model.Events; import com.alibaba.otter.canal.store.model.Events;
import com.google.common.base.Function; import com.google.common.base.Function;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import com.google.common.collect.MapMaker;
import com.google.common.collect.Maps; import com.google.common.collect.Maps;
/** /**
@@ -37,9 +37,10 @@ import com.google.common.collect.Maps;
* @author zebin.xuzb * @author zebin.xuzb
* @version 1.0.0 * @version 1.0.0
*/ */
public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements CanalServer { public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, com.alibaba.otter.canal.server.CanalService
{
private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbeded.class); private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class);
private Map<String, CanalInstance> canalInstances; private Map<String, CanalInstance> canalInstances;
// private Map<ClientIdentity, Position> lastRollbackPostions; // private Map<ClientIdentity, Position> lastRollbackPostions;
private CanalInstanceGenerator canalInstanceGenerator; private CanalInstanceGenerator canalInstanceGenerator;
@@ -47,11 +48,12 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
public void start() { public void start() {
super.start(); super.start();
canalInstances = new MapMaker().makeComputingMap(new Function<String, CanalInstance>() { canalInstances = MigrateMap.makeComputingMap(new Function<String, CanalInstance>()
{
public CanalInstance apply(String destination) { public CanalInstance apply(String destination)
CanalInstance canalInstance = canalInstanceGenerator.generate(destination); {
return canalInstance; return canalInstanceGenerator.generate(destination);
} }
}); });
@@ -74,7 +76,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error(String.format("stop cannalInstance[%s] has an error", entry.getKey()), e); logger.error(String.format("stop CanalInstance[%s] has an error", entry.getKey()), e);
} }
} }
} }
@@ -114,6 +116,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
/** /**
* 客户端订阅，重复订阅时会更新对应的filter信息 * 客户端订阅，重复订阅时会更新对应的filter信息
*/ */
@Override
public void subscribe(ClientIdentity clientIdentity) throws CanalServerException { public void subscribe(ClientIdentity clientIdentity) throws CanalServerException {
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
if (!canalInstance.getMetaManager().isStart()) { if (!canalInstance.getMetaManager().isStart()) {
@@ -140,6 +143,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
/** /**
* 取消订阅 * 取消订阅
*/ */
@Override
public void unsubscribe(ClientIdentity clientIdentity) throws CanalServerException { public void unsubscribe(ClientIdentity clientIdentity) throws CanalServerException {
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
canalInstance.getMetaManager().unsubscribe(clientIdentity); // 执行一下meta订阅 canalInstance.getMetaManager().unsubscribe(clientIdentity); // 执行一下meta订阅
@@ -162,6 +166,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre> * </pre>
*/ */
@Override
public Message get(ClientIdentity clientIdentity, int batchSize) throws CanalServerException { public Message get(ClientIdentity clientIdentity, int batchSize) throws CanalServerException {
return get(clientIdentity, batchSize, null, null); return get(clientIdentity, batchSize, null, null);
} }
@@ -179,6 +184,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre> * </pre>
*/ */
@Override
public Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit) public Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
throws CanalServerException { throws CanalServerException {
checkStart(clientIdentity.getDestination()); checkStart(clientIdentity.getDestination());
@@ -213,8 +219,8 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
}); });
logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]", logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
new Object[] { clientIdentity.getClientId(), batchSize, entrys.size(), batchId, clientIdentity.getClientId(), batchSize, entrys.size(), batchId,
events.getPositionRange() }); events.getPositionRange());
// 直接提交ack // 直接提交ack
ack(clientIdentity, batchId); ack(clientIdentity, batchId);
return new Message(batchId, entrys); return new Message(batchId, entrys);
@@ -230,6 +236,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre> * </pre>
*/ */
@Override
public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException { public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException {
return getWithoutAck(clientIdentity, batchSize, null, null); return getWithoutAck(clientIdentity, batchSize, null, null);
} }
@@ -248,6 +255,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
* 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre> * </pre>
*/ */
@Override
public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit) public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit)
throws CanalServerException { throws CanalServerException {
checkStart(clientIdentity.getDestination()); checkStart(clientIdentity.getDestination());
@@ -285,8 +293,8 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
}); });
logger.info("getWithoutAck successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]", logger.info("getWithoutAck successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
new Object[] { clientIdentity.getClientId(), batchSize, entrys.size(), batchId, clientIdentity.getClientId(), batchSize, entrys.size(), batchId,
events.getPositionRange() }); events.getPositionRange());
return new Message(batchId, entrys); return new Message(batchId, entrys);
} }
@@ -314,6 +322,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
* 注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证) * 注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证)
* </pre> * </pre>
*/ */
@Override
public void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException { public void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException {
checkStart(clientIdentity.getDestination()); checkStart(clientIdentity.getDestination());
checkSubscribe(clientIdentity); checkSubscribe(clientIdentity);
@@ -348,7 +357,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
if (positionRanges.getAck() != null) { if (positionRanges.getAck() != null) {
canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck()); canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
logger.info("ack successfully, clientId:{} batchId:{} position:{}", logger.info("ack successfully, clientId:{} batchId:{} position:{}",
new Object[] { clientIdentity.getClientId(), batchId, positionRanges }); clientIdentity.getClientId(), batchId, positionRanges);
} }
// 可定时清理数据 // 可定时清理数据
@@ -357,8 +366,9 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
} }
/** /**
* 回滚到未进行 {@link ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link ack} 的地方开始拿 * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
*/ */
@Override
public void rollback(ClientIdentity clientIdentity) throws CanalServerException { public void rollback(ClientIdentity clientIdentity) throws CanalServerException {
checkStart(clientIdentity.getDestination()); checkStart(clientIdentity.getDestination());
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
@@ -378,8 +388,9 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
} }
/** /**
* 回滚到未进行 {@link ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link ack} 的地方开始拿 * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
*/ */
@Override
public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException { public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException {
checkStart(clientIdentity.getDestination()); checkStart(clientIdentity.getDestination());
CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination());
@@ -405,7 +416,7 @@ public class CanalServerWithEmbeded extends AbstractCanalLifeCycle implements Ca
canalInstance.getEventStore().rollback();// rollback canalInstance.getEventStore().rollback();// rollback
// eventStore中的状态信息 // eventStore中的状态信息
logger.info("rollback successfully, clientId:{} batchId:{} position:{}", logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
new Object[] { clientIdentity.getClientId(), batchId, positionRanges }); clientIdentity.getClientId(), batchId, positionRanges);
} }
} }

24
server/src/main/java/com/alibaba/otter/canal/server/netty/CanalServerWithNetty.java
View File

@@ -13,7 +13,7 @@ import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import com.alibaba.otter.canal.common.AbstractCanalLifeCycle; import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
import com.alibaba.otter.canal.server.CanalServer; import com.alibaba.otter.canal.server.CanalServer;
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.netty.handler.ClientAuthenticationHandler; import com.alibaba.otter.canal.server.netty.handler.ClientAuthenticationHandler;
import com.alibaba.otter.canal.server.netty.handler.FixedHeaderFrameDecoder; import com.alibaba.otter.canal.server.netty.handler.FixedHeaderFrameDecoder;
import com.alibaba.otter.canal.server.netty.handler.HandshakeInitializationHandler; import com.alibaba.otter.canal.server.netty.handler.HandshakeInitializationHandler;
@@ -27,7 +27,7 @@ import com.alibaba.otter.canal.server.netty.handler.SessionHandler;
*/ */
public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer { public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer {
private CanalServerWithEmbeded embededServer; // 嵌入式server private CanalServerWithEmbedded embeddedServer; // 嵌入式server
private String ip; private String ip;
private int port; private int port;
private Channel serverChannel = null; private Channel serverChannel = null;
@@ -36,15 +36,15 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
public CanalServerWithNetty(){ public CanalServerWithNetty(){
} }
public CanalServerWithNetty(CanalServerWithEmbeded embededServer){ public CanalServerWithNetty(CanalServerWithEmbedded embeddedServer){
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
public void start() { public void start() {
super.start(); super.start();
if (!embededServer.isStart()) { if (!embeddedServer.isStart()) {
embededServer.start(); embeddedServer.start();
} }
this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
@@ -58,9 +58,9 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder()); pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler()); pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler());
pipelines.addLast(ClientAuthenticationHandler.class.getName(), pipelines.addLast(ClientAuthenticationHandler.class.getName(),
new ClientAuthenticationHandler(embededServer)); new ClientAuthenticationHandler(embeddedServer));
SessionHandler sessionHandler = new SessionHandler(embededServer); SessionHandler sessionHandler = new SessionHandler(embeddedServer);
pipelines.addLast(SessionHandler.class.getName(), sessionHandler); pipelines.addLast(SessionHandler.class.getName(), sessionHandler);
return pipelines; return pipelines;
} }
@@ -85,8 +85,8 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
this.bootstrap.releaseExternalResources(); this.bootstrap.releaseExternalResources();
} }
if (embededServer.isStart()) { if (embeddedServer.isStart()) {
embededServer.stop(); embeddedServer.stop();
} }
} }
@@ -98,8 +98,8 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
this.port = port; this.port = port;
} }
public void setEmbededServer(CanalServerWithEmbeded embededServer) { public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
} }

19
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/ClientAuthenticationHandler.java
View File

@@ -19,7 +19,7 @@ import com.alibaba.otter.canal.common.zookeeper.running.ServerRunningMonitors;
import com.alibaba.otter.canal.protocol.CanalPacket.ClientAuth; import com.alibaba.otter.canal.protocol.CanalPacket.ClientAuth;
import com.alibaba.otter.canal.protocol.CanalPacket.Packet; import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.netty.NettyUtils; import com.alibaba.otter.canal.server.netty.NettyUtils;
/** /**
@@ -33,14 +33,14 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class); private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class);
private final int SUPPORTED_VERSION = 3; private final int SUPPORTED_VERSION = 3;
private final int defaultSubscriptorDisconnectIdleTimeout = 5 * 60 * 1000; private final int defaultSubscriptorDisconnectIdleTimeout = 5 * 60 * 1000;
private CanalServerWithEmbeded embededServer; private CanalServerWithEmbedded embeddedServer;
public ClientAuthenticationHandler(){ public ClientAuthenticationHandler(){
} }
public ClientAuthenticationHandler(CanalServerWithEmbeded embededServer){ public ClientAuthenticationHandler(CanalServerWithEmbedded embeddedServer){
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception { public void messageReceived(final ChannelHandlerContext ctx, MessageEvent e) throws Exception {
@@ -58,11 +58,12 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
clientAuth.getFilter()); clientAuth.getFilter());
try { try {
MDC.put("destination", clientIdentity.getDestination()); MDC.put("destination", clientIdentity.getDestination());
embededServer.subscribe(clientIdentity); embeddedServer.subscribe(clientIdentity);
ctx.setAttachment(clientIdentity);// 设置状态数据 ctx.setAttachment(clientIdentity);// 设置状态数据
// 尝试启动，如果已经启动，忽略 // 尝试启动，如果已经启动，忽略
if (!embededServer.isStart(clientIdentity.getDestination())) { if (!embeddedServer.isStart(clientIdentity.getDestination())) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity
.getDestination());
if (!runningMonitor.isStart()) { if (!runningMonitor.isStart()) {
runningMonitor.start(); runningMonitor.start();
} }
@@ -114,8 +115,8 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
} }
} }
public void setEmbededServer(CanalServerWithEmbeded embededServer) { public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
} }

38
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java
View File

@@ -31,7 +31,7 @@ import com.alibaba.otter.canal.protocol.CanalPacket.Sub;
import com.alibaba.otter.canal.protocol.CanalPacket.Unsub; import com.alibaba.otter.canal.protocol.CanalPacket.Unsub;
import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.netty.NettyUtils; import com.alibaba.otter.canal.server.netty.NettyUtils;
/** /**
@@ -43,14 +43,12 @@ import com.alibaba.otter.canal.server.netty.NettyUtils;
public class SessionHandler extends SimpleChannelHandler { public class SessionHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class); private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class);
private CanalServerWithEmbeded embededServer; private CanalServerWithEmbedded embeddedServer;
public SessionHandler(){ public SessionHandler(){ }
} public SessionHandler(CanalServerWithEmbedded embeddedServer){
this.embeddedServer = embeddedServer;
public SessionHandler(CanalServerWithEmbeded embededServer){
this.embededServer = embededServer;
} }
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
@@ -66,10 +64,10 @@ public class SessionHandler extends SimpleChannelHandler {
clientIdentity = new ClientIdentity(sub.getDestination(), Short.valueOf(sub.getClientId()), clientIdentity = new ClientIdentity(sub.getDestination(), Short.valueOf(sub.getClientId()),
sub.getFilter()); sub.getFilter());
MDC.put("destination", clientIdentity.getDestination()); MDC.put("destination", clientIdentity.getDestination());
embededServer.subscribe(clientIdentity); embeddedServer.subscribe(clientIdentity);
// 尝试启动，如果已经启动，忽略 // 尝试启动，如果已经启动，忽略
if (!embededServer.isStart(clientIdentity.getDestination())) { if (!embeddedServer.isStart(clientIdentity.getDestination())) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (!runningMonitor.isStart()) { if (!runningMonitor.isStart()) {
runningMonitor.start(); runningMonitor.start();
@@ -90,7 +88,7 @@ public class SessionHandler extends SimpleChannelHandler {
clientIdentity = new ClientIdentity(unsub.getDestination(), Short.valueOf(unsub.getClientId()), clientIdentity = new ClientIdentity(unsub.getDestination(), Short.valueOf(unsub.getClientId()),
unsub.getFilter()); unsub.getFilter());
MDC.put("destination", clientIdentity.getDestination()); MDC.put("destination", clientIdentity.getDestination());
embededServer.unsubscribe(clientIdentity); embeddedServer.unsubscribe(clientIdentity);
stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭 stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭
NettyUtils.ack(ctx.getChannel(), null); NettyUtils.ack(ctx.getChannel(), null);
} else { } else {
@@ -108,17 +106,17 @@ public class SessionHandler extends SimpleChannelHandler {
// if (get.getAutoAck()) { // if (get.getAutoAck()) {
// if (get.getTimeout() == -1) {//是否是初始值 // if (get.getTimeout() == -1) {//是否是初始值
// message = embededServer.get(clientIdentity, get.getFetchSize()); // message = embeddedServer.get(clientIdentity, get.getFetchSize());
// } else { // } else {
// TimeUnit unit = convertTimeUnit(get.getUnit()); // TimeUnit unit = convertTimeUnit(get.getUnit());
// message = embededServer.get(clientIdentity, get.getFetchSize(), get.getTimeout(), unit); // message = embeddedServer.get(clientIdentity, get.getFetchSize(), get.getTimeout(), unit);
// } // }
// } else { // } else {
if (get.getTimeout() == -1) {//是否是初始值 if (get.getTimeout() == -1) {//是否是初始值
message = embededServer.getWithoutAck(clientIdentity, get.getFetchSize()); message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());
} else { } else {
TimeUnit unit = convertTimeUnit(get.getUnit()); TimeUnit unit = convertTimeUnit(get.getUnit());
message = embededServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(), message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(),
unit); unit);
} }
// } // }
@@ -153,7 +151,7 @@ public class SessionHandler extends SimpleChannelHandler {
// donothing // donothing
} else { } else {
clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId())); clientIdentity = new ClientIdentity(ack.getDestination(), Short.valueOf(ack.getClientId()));
embededServer.ack(clientIdentity, ack.getBatchId()); embeddedServer.ack(clientIdentity, ack.getBatchId());
} }
} else { } else {
NettyUtils.error(401, NettyUtils.error(401,
@@ -169,9 +167,9 @@ public class SessionHandler extends SimpleChannelHandler {
clientIdentity = new ClientIdentity(rollback.getDestination(), clientIdentity = new ClientIdentity(rollback.getDestination(),
Short.valueOf(rollback.getClientId())); Short.valueOf(rollback.getClientId()));
if (rollback.getBatchId() == 0L) { if (rollback.getBatchId() == 0L) {
embededServer.rollback(clientIdentity);// 回滚所有批次 embeddedServer.rollback(clientIdentity);// 回滚所有批次
} else { } else {
embededServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次 embeddedServer.rollback(clientIdentity, rollback.getBatchId()); // 只回滚单个批次
} }
} else { } else {
NettyUtils.error(401, NettyUtils.error(401,
@@ -212,7 +210,7 @@ public class SessionHandler extends SimpleChannelHandler {
} }
private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) { private void stopCanalInstanceIfNecessary(ClientIdentity clientIdentity) {
List<ClientIdentity> clientIdentitys = embededServer.listAllSubscribe(clientIdentity.getDestination()); List<ClientIdentity> clientIdentitys = embeddedServer.listAllSubscribe(clientIdentity.getDestination());
if (clientIdentitys != null && clientIdentitys.size() == 1 && clientIdentitys.contains(clientIdentity)) { if (clientIdentitys != null && clientIdentitys.size() == 1 && clientIdentitys.contains(clientIdentity)) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination()); ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (runningMonitor.isStart()) { if (runningMonitor.isStart()) {
@@ -242,8 +240,8 @@ public class SessionHandler extends SimpleChannelHandler {
} }
} }
public void setEmbededServer(CanalServerWithEmbeded embededServer) { public void setEmbeddedServer(CanalServerWithEmbedded embeddedServer) {
this.embededServer = embededServer; this.embeddedServer = embeddedServer;
} }
} }

6
server/src/test/java/com/alibaba/otter/canal/server/BaseCanalServerWithEmbededTest.java
View File

@@ -14,7 +14,7 @@ import com.alibaba.otter.canal.parse.CanalEventParser;
import com.alibaba.otter.canal.parse.CanalHASwitchable; import com.alibaba.otter.canal.parse.CanalHASwitchable;
import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
public abstract class BaseCanalServerWithEmbededTest { public abstract class BaseCanalServerWithEmbededTest {
@@ -26,12 +26,12 @@ public abstract class BaseCanalServerWithEmbededTest {
protected static final String PASSWORD = "retl"; protected static final String PASSWORD = "retl";
protected static final String FILTER = "retl\\..*,erosa.zk_complaint_bizdata"; protected static final String FILTER = "retl\\..*,erosa.zk_complaint_bizdata";
private CanalServerWithEmbeded server; private CanalServerWithEmbedded server;
private ClientIdentity clientIdentity = new ClientIdentity(DESTINATION, (short) 1); ; private ClientIdentity clientIdentity = new ClientIdentity(DESTINATION, (short) 1); ;
@Before @Before
public void setUp() { public void setUp() {
server = new CanalServerWithEmbeded(); server = new CanalServerWithEmbedded();
server.setCanalInstanceGenerator(new CanalInstanceGenerator() { server.setCanalInstanceGenerator(new CanalInstanceGenerator() {
public CanalInstance generate(String destination) { public CanalInstance generate(String destination) {

2
server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithEmbeded_StandaloneTest.java → server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithEmbedded_StandaloneTest.java
View File

@@ -11,7 +11,7 @@ import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType; import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode; import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
public class CanalServerWithEmbeded_StandaloneTest extends BaseCanalServerWithEmbededTest { public class CanalServerWithEmbedded_StandaloneTest extends BaseCanalServerWithEmbededTest {
protected Canal buildCanal() { protected Canal buildCanal() {
Canal canal = new Canal(); Canal canal = new Canal();

2
server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithEmbeded_StandbyTest.java → server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithEmbedded_StandbyTest.java
View File

@@ -15,7 +15,7 @@ import com.alibaba.otter.canal.instance.manager.model.CanalParameter.MetaMode;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType; import com.alibaba.otter.canal.instance.manager.model.CanalParameter.SourcingType;
import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode; import com.alibaba.otter.canal.instance.manager.model.CanalParameter.StorageMode;
public class CanalServerWithEmbeded_StandbyTest extends BaseCanalServerWithEmbededTest { public class CanalServerWithEmbedded_StandbyTest extends BaseCanalServerWithEmbededTest {
private ZkClient zkClient = new ZkClient(cluster1); private ZkClient zkClient = new ZkClient(cluster1);

12
server/src/test/java/com/alibaba/otter/canal/server/CanalServerWithNettyTest.java
View File

@@ -31,7 +31,7 @@ import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType; import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
import com.alibaba.otter.canal.protocol.CanalPacket.Sub; import com.alibaba.otter.canal.protocol.CanalPacket.Sub;
import com.alibaba.otter.canal.protocol.CanalPacket.Unsub; import com.alibaba.otter.canal.protocol.CanalPacket.Unsub;
import com.alibaba.otter.canal.server.embeded.CanalServerWithEmbeded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.server.netty.CanalServerWithNetty; import com.alibaba.otter.canal.server.netty.CanalServerWithNetty;
public class CanalServerWithNettyTest { public class CanalServerWithNettyTest {
@@ -49,16 +49,18 @@ public class CanalServerWithNettyTest {
@Before @Before
public void setUp() { public void setUp() {
CanalServerWithEmbeded embededServer = new CanalServerWithEmbeded(); CanalServerWithEmbedded embeddedServer = new CanalServerWithEmbedded();
embededServer.setCanalInstanceGenerator(new CanalInstanceGenerator() { embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator()
{
public CanalInstance generate(String destination) { public CanalInstance generate(String destination)
{
Canal canal = buildCanal(); Canal canal = buildCanal();
return new CanalInstanceWithManager(canal, FILTER); return new CanalInstanceWithManager(canal, FILTER);
} }
}); });
nettyServer = new CanalServerWithNetty(embededServer); nettyServer = new CanalServerWithNetty(embeddedServer);
nettyServer.setPort(1088); nettyServer.setPort(1088);
nettyServer.start(); nettyServer.start();
} }

8
sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java
View File

@@ -132,8 +132,8 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
boolean need = filter.filter(name); boolean need = filter.filter(name);
if (!need) { if (!need) {
logger.debug("filter name[{}] entry : {}:{}", logger.debug("filter name[{}] entry : {}:{}",
new Object[] { name, event.getEntry().getHeader().getLogfileName(), name, event.getEntry().getHeader().getLogfileName(),
event.getEntry().getHeader().getLogfileOffset() }); event.getEntry().getHeader().getLogfileOffset());
} }
return need; return need;
@@ -178,9 +178,7 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
} }
private String getSchemaNameAndTableName(CanalEntry.Entry entry) { private String getSchemaNameAndTableName(CanalEntry.Entry entry) {
StringBuilder result = new StringBuilder(); return entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName();
result.append(entry.getHeader().getSchemaName()).append(".").append(entry.getHeader().getTableName());
return result.toString();
} }
public void setEventStore(CanalEventStore<Event> eventStore) { public void setEventStore(CanalEventStore<Event> eventStore) {