diff --git a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java index b8f84395..fb0db67f 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java @@ -19,14 +19,14 @@ public interface CanalConnector { * * @throws CanalClientException */ - public void connect() throws CanalClientException; + void connect() throws CanalClientException; /** * 释放链接 * * @throws CanalClientException */ - public void disconnect() throws CanalClientException; + void disconnect() throws CanalClientException; /** * 检查下链接是否合法 @@ -43,7 +43,7 @@ public interface CanalConnector { * * @throws CanalClientException */ - public boolean checkValid() throws CanalClientException; + boolean checkValid() throws CanalClientException; /** * 客户端订阅，重复订阅时会更新对应的filter信息 @@ -56,7 +56,6 @@ public interface CanalConnector { * TODO: 后续可以考虑，如果本次提交的filter不为空，在执行过滤时，是对canal server filter + 本次filter的交集处理，达到只取1份binlog数据，多个客户端消费不同的表 * * - * @param clientIdentity * @throws CanalClientException */ void subscribe(String filter) throws CanalClientException; @@ -64,7 +63,6 @@ public interface CanalConnector { /** * 客户端订阅，不提交客户端filter，以服务端的filter为准 * - * @param clientIdentity * @throws CanalClientException */ void subscribe() throws CanalClientException; @@ -72,7 +70,6 @@ public interface CanalConnector { /** * 取消订阅 * - * @param clientIdentity * @throws CanalClientException */ void unsubscribe() throws CanalClientException; @@ -140,14 +137,14 @@ public interface CanalConnector { void ack(long batchId) throws CanalClientException; /** - * 回滚到未进行 {@link ack} 的地方，指定回滚具体的batchId + * 回滚到未进行 {@link #ack} 的地方，指定回滚具体的batchId * * @throws CanalClientException */ void rollback(long batchId) throws CanalClientException; /** - * 回滚到未进行 {@link ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link ack} 的地方开始拿 + * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿 * * @throws CanalClientException */ diff --git a/common/pom.xml b/common/pom.xml index 620ed474..b51d6961 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -6,7 +6,6 @@ 1.0.20-SNAPSHOT ../pom.xml - com.alibaba.otter canal.common jar canal common module for otter ${project.version} diff --git a/instance/core/pom.xml b/instance/core/pom.xml index b4bb5428..2c53ef79 100644 --- a/instance/core/pom.xml +++ b/instance/core/pom.xml @@ -6,7 +6,6 @@ 1.0.20-SNAPSHOT ../../pom.xml - com.alibaba.otter canal.instance.core jar canal instance core module for otter ${project.version} diff --git a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java index 0d9c0d32..e5b84d1e 100644 --- a/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java +++ b/instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java @@ -89,8 +89,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca this.destination = canal.getName(); this.filter = filter; - logger.info("init CannalInstance for {}-{} with parameters:{}", - new Object[] { canalId, destination, parameters }); + logger.info("init CannalInstance for {}-{} with parameters:{}", canalId, destination, parameters); // 初始化报警机制 initAlarmHandler(); // 初始化metaManager @@ -116,8 +115,8 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca public void start() { super.start(); // 初始化metaManager - logger.info("start CannalInstance for {}-{} with parameters:{}", new Object[] { canalId, destination, - parameters }); + logger.info("start CannalInstance for {}-{} with parameters:{}", canalId, destination, + parameters); if (!metaManager.isStart()) { metaManager.start(); @@ -336,7 +335,7 @@ public class CanalInstanceWithManager extends CanalInstanceSupport implements Ca } private CanalEventParser doInitEventParser(SourcingType type, List dbAddresses) { - CanalEventParser eventParser = null; + CanalEventParser eventParser; if (type.isMysql()) { MysqlEventParser mysqlEventParser = new MysqlEventParser(); mysqlEventParser.setDestination(destination); diff --git a/meta/src/main/java/com/alibaba/otter/canal/meta/CanalMetaManager.java b/meta/src/main/java/com/alibaba/otter/canal/meta/CanalMetaManager.java index 534e6d4a..17faf395 100644 --- a/meta/src/main/java/com/alibaba/otter/canal/meta/CanalMetaManager.java +++ b/meta/src/main/java/com/alibaba/otter/canal/meta/CanalMetaManager.java @@ -35,12 +35,12 @@ public interface CanalMetaManager extends CanalLifeCycle { void unsubscribe(ClientIdentity clientIdentity) throws CanalMetaManagerException; /** - * 获取cuosr游标 + * 获取 cursor 游标 */ Position getCursor(ClientIdentity clientIdentity) throws CanalMetaManagerException; /** - * 更新cuosr游标 + * 更新 cursor 游标 */ void updateCursor(ClientIdentity clientIdentity, Position position) throws CanalMetaManagerException; diff --git a/parse/pom.xml b/parse/pom.xml index 7ce7a62c..5aad44bd 100644 --- a/parse/pom.xml +++ b/parse/pom.xml @@ -6,7 +6,6 @@ 1.0.20-SNAPSHOT ../pom.xml - com.alibaba.otter canal.parse jar canal parse module for otter ${project.version} diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/ClientIdentity.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/ClientIdentity.java index 52e3fb05..1eccdef4 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/ClientIdentity.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/ClientIdentity.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.protocol; +import java.io.Serializable; import org.apache.commons.lang.StringUtils; 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 * @version 1.0.0 */ -public class ClientIdentity { +public class ClientIdentity implements Serializable +{ private String destination; private short clientId; diff --git a/server/pom.xml b/server/pom.xml index 27cbb50a..eb2abcc7 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -6,7 +6,6 @@ 1.0.20-SNAPSHOT ../pom.xml - com.alibaba.otter canal.server jar canal server module for otter ${project.version} diff --git a/server/src/main/java/com/alibaba/otter/canal/server/CanalService.java b/server/src/main/java/com/alibaba/otter/canal/server/CanalService.java new file mode 100644 index 00000000..48709ab8 --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/server/CanalService.java @@ -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; +} diff --git a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java index c23ff139..38ea14d8 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java @@ -28,7 +28,6 @@ import com.alibaba.otter.canal.store.model.Event; import com.alibaba.otter.canal.store.model.Events; import com.google.common.base.Function; import com.google.common.collect.Lists; -import com.google.common.collect.MapMaker; import com.google.common.collect.Maps; /** @@ -38,7 +37,8 @@ import com.google.common.collect.Maps; * @author zebin.xuzb * @version 1.0.0 */ -public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer { +public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, com.alibaba.otter.canal.server.CanalService +{ private static final Logger logger = LoggerFactory.getLogger(CanalServerWithEmbedded.class); private Map canalInstances; @@ -116,6 +116,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C /** * 客户端订阅，重复订阅时会更新对应的filter信息 */ + @Override public void subscribe(ClientIdentity clientIdentity) throws CanalServerException { CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); if (!canalInstance.getMetaManager().isStart()) { @@ -142,6 +143,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C /** * 取消订阅 */ + @Override public void unsubscribe(ClientIdentity clientIdentity) throws CanalServerException { CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); canalInstance.getMetaManager().unsubscribe(clientIdentity); // 执行一下meta订阅 @@ -164,6 +166,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * */ + @Override public Message get(ClientIdentity clientIdentity, int batchSize) throws CanalServerException { return get(clientIdentity, batchSize, null, null); } @@ -181,6 +184,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * */ + @Override public Message get(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit) throws CanalServerException { checkStart(clientIdentity.getDestination()); @@ -215,8 +219,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C }); logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]", - new Object[] { clientIdentity.getClientId(), batchSize, entrys.size(), batchId, - events.getPositionRange() }); + clientIdentity.getClientId(), batchSize, entrys.size(), batchId, + events.getPositionRange()); // 直接提交ack ack(clientIdentity, batchId); return new Message(batchId, entrys); @@ -232,6 +236,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * */ + @Override public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize) throws CanalServerException { return getWithoutAck(clientIdentity, batchSize, null, null); } @@ -250,6 +255,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题) * */ + @Override public Message getWithoutAck(ClientIdentity clientIdentity, int batchSize, Long timeout, TimeUnit unit) throws CanalServerException { checkStart(clientIdentity.getDestination()); @@ -287,8 +293,8 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C }); logger.info("getWithoutAck successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]", - new Object[] { clientIdentity.getClientId(), batchSize, entrys.size(), batchId, - events.getPositionRange() }); + clientIdentity.getClientId(), batchSize, entrys.size(), batchId, + events.getPositionRange()); return new Message(batchId, entrys); } @@ -316,6 +322,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C * 注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证) * */ + @Override public void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException { checkStart(clientIdentity.getDestination()); checkSubscribe(clientIdentity); @@ -350,7 +357,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C if (positionRanges.getAck() != null) { canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck()); logger.info("ack successfully, clientId:{} batchId:{} position:{}", - new Object[] { clientIdentity.getClientId(), batchId, positionRanges }); + clientIdentity.getClientId(), batchId, positionRanges); } // 可定时清理数据 @@ -361,6 +368,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C /** * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿 */ + @Override public void rollback(ClientIdentity clientIdentity) throws CanalServerException { checkStart(clientIdentity.getDestination()); CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); @@ -382,6 +390,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C /** * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿 */ + @Override public void rollback(ClientIdentity clientIdentity, Long batchId) throws CanalServerException { checkStart(clientIdentity.getDestination()); CanalInstance canalInstance = canalInstances.get(clientIdentity.getDestination()); @@ -407,7 +416,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C canalInstance.getEventStore().rollback();// rollback // eventStore中的状态信息 logger.info("rollback successfully, clientId:{} batchId:{} position:{}", - new Object[] { clientIdentity.getClientId(), batchId, positionRanges }); + clientIdentity.getClientId(), batchId, positionRanges); } } diff --git a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java index 8474950e..1e85eaaa 100644 --- a/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java +++ b/sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java @@ -132,8 +132,8 @@ public class EntryEventSink extends AbstractCanalEventSink eventStore) {