wyl/canal
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

cleanup code, extract CanalService

Browse Source
This commit is contained in:
wener
2015-04-27 14:57:01 +08:00
parent fc9b616f4a
commit 596b060649
11 changed files with 63 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java
+5 -8
View File
@@ -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数据，多个客户端消费不同的表
* </pre>
*
* @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
*/
common/pom.xml
-1
View File
@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.common</artifactId>
<packaging>jar</packaging>
<name>canal common module for otter ${project.version}</name>
instance/core/pom.xml
-1
View File
@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version>
<relativePath>../../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.instance.core</artifactId>
<packaging>jar</packaging>
<name>canal instance core module for otter ${project.version}</name>
instance/manager/src/main/java/com/alibaba/otter/canal/instance/manager/CanalInstanceWithManager.java
+4 -5
View File
@@ -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<InetSocketAddress> dbAddresses) {
CanalEventParser eventParser = null;
CanalEventParser eventParser;
if (type.isMysql()) {
MysqlEventParser mysqlEventParser = new MysqlEventParser();
mysqlEventParser.setDestination(destination);
meta/src/main/java/com/alibaba/otter/canal/meta/CanalMetaManager.java
+2 -2
View File
@@ -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;
parse/pom.xml
-1
View File
@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.parse</artifactId>
<packaging>jar</packaging>
<name>canal parse module for otter ${project.version}</name>
protocol/src/main/java/com/alibaba/otter/canal/protocol/ClientIdentity.java
+3 -1
View File
@@ -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;
server/pom.xml
-1
View File
@@ -6,7 +6,6 @@
<version>1.0.20-SNAPSHOT</version>
<relativePath>../pom.xml</relativePath>
</parent>
<groupId>com.alibaba.otter</groupId>
<artifactId>canal.server</artifactId>
<packaging>jar</packaging>
<name>canal server module for otter ${project.version}</name>
server/src/main/java/com/alibaba/otter/canal/server/CanalService.java
+29
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;
}
server/src/main/java/com/alibaba/otter/canal/server/embedded/CanalServerWithEmbedded.java
+17 -8
View File
@@ -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<String, CanalInstance> 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，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre>
*/
@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，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre>
*/
@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，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre>
*/
@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，拿到第二批数据，这样就会导致数据顺序性出现问题)
* </pre>
*/
@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(需有客户端保证)
* </pre>
*/
@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);
}
}
sink/src/main/java/com/alibaba/otter/canal/sink/entry/EntryEventSink.java
+3 -5
View File
@@ -132,8 +132,8 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
boolean need = filter.filter(name);
if (!need) {
logger.debug("filter name[{}] entry : {}:{}",
new Object[] { name, event.getEntry().getHeader().getLogfileName(),
event.getEntry().getHeader().getLogfileOffset() });
name, event.getEntry().getHeader().getLogfileName(),
event.getEntry().getHeader().getLogfileOffset());
}
return need;
@@ -178,9 +178,7 @@ public class EntryEventSink extends AbstractCanalEventSink<List<CanalEntry.Entry
}
private String getSchemaNameAndTableName(CanalEntry.Entry entry) {
StringBuilder result = new StringBuilder();
result.append(entry.getHeader().getSchemaName()).append(".").append(entry.getHeader().getTableName());
return result.toString();
return entry.getHeader().getSchemaName() + "." + entry.getHeader().getTableName();
}
public void setEventStore(CanalEventStore<Event> eventStore) {