diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java index d1e81ec0..0b112757 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStater.java @@ -52,6 +52,8 @@ public class CanalStater { // disable netty System.setProperty(CanalConstants.CANAL_WITHOUT_NETTY, "true"); String autoScan = CanalController.getProperty(properties, CanalConstants.CANAL_AUTO_SCAN); + // 设置为raw避免ByteString->Entry的二次解析 + System.setProperty("canal.instance.memory.rawEntry", "false"); if ("true".equals(autoScan)) { String rootDir = CanalController.getProperty(properties, CanalConstants.CANAL_CONF_DIR); if (StringUtils.isEmpty(rootDir)) { @@ -68,13 +70,13 @@ public class CanalStater { }); if (instanceDirs != null && instanceDirs.length > 0) { List instances = Lists.transform(Arrays.asList(instanceDirs), - new Function() { + new Function() { - @Override - public String apply(File instanceDir) { - return instanceDir.getName(); - } - }); + @Override + public String apply(File instanceDir) { + return instanceDir.getName(); + } + }); System.setProperty(CanalConstants.CANAL_DESTINATIONS, Joiner.on(",").join(instances)); } } diff --git a/deployer/src/main/resources/example/instance.properties b/deployer/src/main/resources/example/instance.properties index 65a8c6d2..bc1db836 100644 --- a/deployer/src/main/resources/example/instance.properties +++ b/deployer/src/main/resources/example/instance.properties @@ -46,7 +46,7 @@ canal.instance.filter.black.regex= # mq config canal.mq.topic=example # dynamic topic route by schema or table regex -#canal.mq.dynamicTopic=.*,mytest,mytest\\..*,mytest2.user +#canal.mq.dynamicTopic=mytest1.user,mytest2\\..*,.*\\..* canal.mq.partition=0 # hash partition config #canal.mq.partitionsNum=4 diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java index 370b848d..17568d44 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQMessageUtils.java @@ -509,7 +509,7 @@ public class MQMessageUtils { } private static Set matchTopics(String name, String dynamicTopicConfigs) { - String[] router = StringUtils.split(dynamicTopicConfigs, ';'); + String[] router = StringUtils.split(dynamicTopicConfigs, ','); Set topics = new HashSet<>(); for (String item : router) { int i = item.indexOf(":"); diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java index ae929b41..796a60b5 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java @@ -53,9 +53,6 @@ public class CanalMQStarter { System.setProperty("canal.instance.filter.transaction.entry", "true"); } - // 设置为raw避免ByteString->Entry的二次解析 - System.setProperty("canal.instance.memory.rawEntry", "false"); - canalServer = CanalServerWithEmbedded.instance(); // 对应每个instance启动一个worker线程