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

update guava to 18.0, magriate compute map by package level implement

Browse Source
This commit is contained in:
wener
2015-04-23 22:15:57 +08:00
parent bdf3b0c861
commit 6a0733bbab
14 changed files with 633 additions and 502 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
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,9 +24,11 @@ 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);
} }
}); });

92
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;
@@ -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);
} }
}); });
@@ -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");
} }
} }

7
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,9 +40,11 @@ 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);
} }
}); });

36
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 { {
try
{
PatternCompiler pc = new Perl5Compiler(); PatternCompiler pc = new Perl5Compiler();
return pc.compile( return pc.compile(
pattern, pattern,
Perl5Compiler.CASE_INSENSITIVE_MASK Perl5Compiler.CASE_INSENSITIVE_MASK
| Perl5Compiler.READ_ONLY_MASK | Perl5Compiler.READ_ONLY_MASK
| Perl5Compiler.SINGLELINE_MASK); | Perl5Compiler.SINGLELINE_MASK);
} catch (MalformedPatternException e) { } catch (MalformedPatternException e)
{
throw new CanalFilterException( throw new CanalFilterException(
e); 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();
} }
} }

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();
} }
}); });

134
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;

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;
} }
} }

15
pom.xml
View File

@@ -1,4 +1,5 @@
<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"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion> <modelVersion>4.0.0</modelVersion>
<groupId>com.alibaba.otter</groupId> <groupId>com.alibaba.otter</groupId>
<artifactId>canal</artifactId> <artifactId>canal</artifactId>
@@ -182,7 +183,7 @@
<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>
@@ -208,28 +209,28 @@
<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> <dependency>
<groupId>ch.qos.logback</groupId> <groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId> <artifactId>logback-classic</artifactId>
<version>1.0.6</version> <version>1.1.3</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.slf4j</groupId> <groupId>org.slf4j</groupId>
<artifactId>jcl-over-slf4j</artifactId> <artifactId>jcl-over-slf4j</artifactId>
<version>1.6.0</version> <version>1.7.12</version>
</dependency> </dependency>
<dependency> <dependency>
<groupId>org.slf4j</groupId> <groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId> <artifactId>slf4j-api</artifactId>
<version>1.6.0</version> <version>1.7.12</version>
</dependency> </dependency>
<!-- test dependency --> <!-- test dependency -->
<dependency> <dependency>
<groupId>junit</groupId> <groupId>junit</groupId>
<artifactId>junit</artifactId> <artifactId>junit</artifactId>
<version>4.5</version> <version>4.12</version>
<scope>test</scope> <scope>test</scope>
</dependency> </dependency>
<dependency> <dependency>

7
server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java
View File

@@ -1,5 +1,6 @@
package com.alibaba.otter.canal.server.embedded; 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;
@@ -47,9 +48,11 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
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)
{
return canalInstanceGenerator.generate(destination); return canalInstanceGenerator.generate(destination);
} }
}); });