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

fixed code template and format

Browse Source
This commit is contained in:
agapple
2015-04-28 21:11:41 +08:00
parent ea889cb004
commit 7ebb3fd1ab
86 changed files with 2763 additions and 3554 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
server/src/main/java/com/alibaba/otter/canal/server/CanalService.java
View File

@@ -1,25 +1,25 @@
package com.alibaba.otter.canal.server;
import java.util.concurrent.TimeUnit;
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
{
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 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;
throws CanalServerException;
void ack(ClientIdentity clientIdentity, long batchId) throws CanalServerException;

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

@@ -1,6 +1,5 @@
package com.alibaba.otter.canal.server.embedded;
import com.google.common.collect.MigrateMap;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -29,6 +28,7 @@ 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.Maps;
import com.google.common.collect.MigrateMap;
/**
* 嵌入式版本实现
@@ -37,8 +37,7 @@ import com.google.common.collect.Maps;
* @author zebin.xuzb
* @version 1.0.0
*/
public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements CanalServer, com.alibaba.otter.canal.server.CanalService
{
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;
@@ -48,11 +47,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
public void start() {
super.start();
canalInstances = MigrateMap.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);
}
});
@@ -195,9 +192,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().getLastestBatch(clientIdentity);
if (positionRanges != null) {
throw new CanalServerException(
String.format("clientId:%s has last batch:[%s] isn't ack , maybe loss data",
clientIdentity.getClientId(), positionRanges));
throw new CanalServerException(String.format("clientId:%s has last batch:[%s] isn't ack , maybe loss data",
clientIdentity.getClientId(),
positionRanges));
}
Events<Event> events = null;
@@ -219,8 +216,11 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
});
logger.info("get successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
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);
@@ -293,8 +293,11 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
});
logger.info("getWithoutAck successfully, clientId:{} batchSize:{} real size is {} and result is [batchId:{} , position:{}]",
clientIdentity.getClientId(), batchSize, entrys.size(), batchId,
events.getPositionRange());
clientIdentity.getClientId(),
batchSize,
entrys.size(),
batchId,
events.getPositionRange());
return new Message(batchId, entrys);
}
@@ -331,9 +334,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
PositionRange<LogPosition> positionRanges = null;
positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity, batchId); // 更新位置
if (positionRanges == null) { // 说明是重复的ack/rollback
throw new CanalServerException(
String.format("ack error , clientId:%s batchId:%d is not exist , please check",
clientIdentity.getClientId(), batchId));
throw new CanalServerException(String.format("ack error , clientId:%s batchId:%d is not exist , please check",
clientIdentity.getClientId(),
batchId));
}
// 更新cursor最好严格判断下位置是否有跳跃更新
@@ -357,7 +360,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
if (positionRanges.getAck() != null) {
canalInstance.getMetaManager().updateCursor(clientIdentity, positionRanges.getAck());
logger.info("ack successfully, clientId:{} batchId:{} position:{}",
clientIdentity.getClientId(), batchId, positionRanges);
clientIdentity.getClientId(),
batchId,
positionRanges);
}
// 可定时清理数据
@@ -403,11 +408,11 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
synchronized (canalInstance) {
// 清除batch信息
PositionRange<LogPosition> positionRanges = canalInstance.getMetaManager().removeBatch(clientIdentity,
batchId);
batchId);
if (positionRanges == null) { // 说明是重复的ack/rollback
throw new CanalServerException(
String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
clientIdentity.getClientId(), batchId));
throw new CanalServerException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
clientIdentity.getClientId(),
batchId));
}
// lastRollbackPostions.put(clientIdentity,
@@ -416,7 +421,9 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
canalInstance.getEventStore().rollback();// rollback
// eventStore中的状态信息
logger.info("rollback successfully, clientId:{} batchId:{} position:{}",
clientIdentity.getClientId(), batchId, positionRanges);
clientIdentity.getClientId(),
batchId,
positionRanges);
}
}
@@ -451,7 +458,7 @@ public class CanalServerWithEmbedded extends AbstractCanalLifeCycle implements C
boolean hasSubscribe = canalInstance.getMetaManager().hasSubscribe(clientIdentity);
if (!hasSubscribe) {
throw new CanalServerException(String.format("ClientIdentity:%s should subscribe first",
clientIdentity.toString()));
clientIdentity.toString()));
}
}

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

@@ -27,11 +27,11 @@ import com.alibaba.otter.canal.server.netty.handler.SessionHandler;
*/
public class CanalServerWithNetty extends AbstractCanalLifeCycle implements CanalServer {
private CanalServerWithEmbedded embeddedServer; // 嵌入式server
private String ip;
private int port;
private Channel serverChannel = null;
private ServerBootstrap bootstrap = null;
private CanalServerWithEmbedded embeddedServer; // 嵌入式server
private String ip;
private int port;
private Channel serverChannel = null;
private ServerBootstrap bootstrap = null;
public CanalServerWithNetty(){
}
@@ -48,7 +48,7 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
}
this.bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
Executors.newCachedThreadPool()));
Executors.newCachedThreadPool()));
// 构造对应的pipeline
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@@ -58,7 +58,7 @@ public class CanalServerWithNetty extends AbstractCanalLifeCycle implements Cana
pipelines.addLast(FixedHeaderFrameDecoder.class.getName(), new FixedHeaderFrameDecoder());
pipelines.addLast(HandshakeInitializationHandler.class.getName(), new HandshakeInitializationHandler());
pipelines.addLast(ClientAuthenticationHandler.class.getName(),
new ClientAuthenticationHandler(embeddedServer));
new ClientAuthenticationHandler(embeddedServer));
SessionHandler sessionHandler = new SessionHandler(embeddedServer);
pipelines.addLast(SessionHandler.class.getName(), sessionHandler);

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

@@ -32,10 +32,13 @@ public class NettyUtils {
}
public static void ack(Channel channel, ChannelFutureListener channelFutureListner) {
write(
channel,
Packet.newBuilder().setType(CanalPacket.PacketType.ACK).setBody(Ack.newBuilder().build().toByteString()).build().toByteArray(),
channelFutureListner);
write(channel,
Packet.newBuilder()
.setType(CanalPacket.PacketType.ACK)
.setBody(Ack.newBuilder().build().toByteString())
.build()
.toByteArray(),
channelFutureListner);
}
public static void error(int errorCode, String errorMessage, Channel channel,
@@ -45,11 +48,12 @@ public class NettyUtils {
}
logger.error("ErrotCode:{} , Caused by : \n{}", errorCode, errorMessage);
write(
channel,
Packet.newBuilder().setType(CanalPacket.PacketType.ACK).setBody(
Ack.newBuilder().setErrorCode(errorCode).setErrorMessage(
errorMessage).build().toByteString()).build().toByteArray(),
channelFutureListener);
write(channel,
Packet.newBuilder()
.setType(CanalPacket.PacketType.ACK)
.setBody(Ack.newBuilder().setErrorCode(errorCode).setErrorMessage(errorMessage).build().toByteString())
.build()
.toByteArray(),
channelFutureListener);
}
}

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

@@ -30,9 +30,9 @@ import com.alibaba.otter.canal.server.netty.NettyUtils;
*/
public class ClientAuthenticationHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class);
private final int SUPPORTED_VERSION = 3;
private final int defaultSubscriptorDisconnectIdleTimeout = 5 * 60 * 1000;
private static final Logger logger = LoggerFactory.getLogger(ClientAuthenticationHandler.class);
private final int SUPPORTED_VERSION = 3;
private final int defaultSubscriptorDisconnectIdleTimeout = 5 * 60 * 1000;
private CanalServerWithEmbedded embeddedServer;
public ClientAuthenticationHandler(){
@@ -62,8 +62,7 @@ public class ClientAuthenticationHandler extends SimpleChannelHandler {
ctx.setAttachment(clientIdentity);// 设置状态数据
// 尝试启动，如果已经启动，忽略
if (!embeddedServer.isStart(clientIdentity.getDestination())) {
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity
.getDestination());
ServerRunningMonitor runningMonitor = ServerRunningMonitors.getRunningMonitor(clientIdentity.getDestination());
if (!runningMonitor.isStart()) {
runningMonitor.start();
}

2
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/FixedHeaderFrameDecoder.java
View File

@@ -16,6 +16,6 @@ public class FixedHeaderFrameDecoder extends ReplayingDecoder<VoidEnum> {
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, VoidEnum state)
throws Exception {
return buffer.readBytes(buffer.readInt());
return buffer.readBytes(buffer.readInt());
}
}

7
server/src/main/java/com/alibaba/otter/canal/server/netty/handler/HandshakeInitializationHandler.java
View File

@@ -22,8 +22,11 @@ public class HandshakeInitializationHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(HandshakeInitializationHandler.class);
public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
byte[] body = Packet.newBuilder().setType(CanalPacket.PacketType.HANDSHAKE).setBody(
Handshake.newBuilder().build().toByteString()).build().toByteArray();
byte[] body = Packet.newBuilder()
.setType(CanalPacket.PacketType.HANDSHAKE)
.setBody(Handshake.newBuilder().build().toByteString())
.build()
.toByteArray();
NettyUtils.write(ctx.getChannel(), body, null);
logger.info("send handshake initialization packet to : {}", ctx.getChannel());
}

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

@@ -42,10 +42,11 @@ import com.alibaba.otter.canal.server.netty.NettyUtils;
*/
public class SessionHandler extends SimpleChannelHandler {
private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class);
private static final Logger logger = LoggerFactory.getLogger(SessionHandler.class);
private CanalServerWithEmbedded embeddedServer;
public SessionHandler(){ }
public SessionHandler(){
}
public SessionHandler(CanalServerWithEmbedded embeddedServer){
this.embeddedServer = embeddedServer;
@@ -61,8 +62,9 @@ public class SessionHandler extends SimpleChannelHandler {
case SUBSCRIPTION:
Sub sub = Sub.parseFrom(packet.getBody());
if (StringUtils.isNotEmpty(sub.getDestination()) && StringUtils.isNotEmpty(sub.getClientId())) {
clientIdentity = new ClientIdentity(sub.getDestination(), Short.valueOf(sub.getClientId()),
sub.getFilter());
clientIdentity = new ClientIdentity(sub.getDestination(),
Short.valueOf(sub.getClientId()),
sub.getFilter());
MDC.put("destination", clientIdentity.getDestination());
embeddedServer.subscribe(clientIdentity);
@@ -78,23 +80,26 @@ public class SessionHandler extends SimpleChannelHandler {
NettyUtils.ack(ctx.getChannel(), null);
} else {
NettyUtils.error(401,
MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage(),
ctx.getChannel(), null);
MessageFormatter.format("destination or clientId is null", sub.toString()).getMessage(),
ctx.getChannel(),
null);
}
break;
case UNSUBSCRIPTION:
Unsub unsub = Unsub.parseFrom(packet.getBody());
if (StringUtils.isNotEmpty(unsub.getDestination()) && StringUtils.isNotEmpty(unsub.getClientId())) {
clientIdentity = new ClientIdentity(unsub.getDestination(), Short.valueOf(unsub.getClientId()),
unsub.getFilter());
clientIdentity = new ClientIdentity(unsub.getDestination(),
Short.valueOf(unsub.getClientId()),
unsub.getFilter());
MDC.put("destination", clientIdentity.getDestination());
embeddedServer.unsubscribe(clientIdentity);
stopCanalInstanceIfNecessary(clientIdentity);// 尝试关闭
NettyUtils.ack(ctx.getChannel(), null);
} else {
NettyUtils.error(401,
MessageFormatter.format("destination or clientId is null", unsub.toString()).getMessage(),
ctx.getChannel(), null);
MessageFormatter.format("destination or clientId is null", unsub.toString()).getMessage(),
ctx.getChannel(),
null);
}
break;
case GET:
@@ -104,22 +109,26 @@ public class SessionHandler extends SimpleChannelHandler {
MDC.put("destination", clientIdentity.getDestination());
Message message = null;
// if (get.getAutoAck()) {
// if (get.getTimeout() == -1) {//是否是初始值
// message = embeddedServer.get(clientIdentity, get.getFetchSize());
// } else {
// TimeUnit unit = convertTimeUnit(get.getUnit());
// message = embeddedServer.get(clientIdentity, get.getFetchSize(), get.getTimeout(), unit);
// }
// } else {
if (get.getTimeout() == -1) {//是否是初始值
// if (get.getAutoAck()) {
// if (get.getTimeout() == -1) {//是否是初始值
// message = embeddedServer.get(clientIdentity,
// get.getFetchSize());
// } else {
// TimeUnit unit = convertTimeUnit(get.getUnit());
// message = embeddedServer.get(clientIdentity,
// get.getFetchSize(), get.getTimeout(), unit);
// }
// } else {
if (get.getTimeout() == -1) {// 是否是初始值
message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize());
} else {
TimeUnit unit = convertTimeUnit(get.getUnit());
message = embeddedServer.getWithoutAck(clientIdentity, get.getFetchSize(), get.getTimeout(),
unit);
message = embeddedServer.getWithoutAck(clientIdentity,
get.getFetchSize(),
get.getTimeout(),
unit);
}
// }
// }
Packet.Builder packetBuilder = CanalPacket.Packet.newBuilder();
packetBuilder.setType(PacketType.MESSAGES);
@@ -135,8 +144,9 @@ public class SessionHandler extends SimpleChannelHandler {
NettyUtils.write(ctx.getChannel(), packetBuilder.build().toByteArray(), null);// 输出数据
} else {
NettyUtils.error(401,
MessageFormatter.format("destination or clientId is null", get.toString()).getMessage(),
ctx.getChannel(), null);
MessageFormatter.format("destination or clientId is null", get.toString()).getMessage(),
ctx.getChannel(),
null);
}
break;
case CLIENTACK:
@@ -145,8 +155,9 @@ public class SessionHandler extends SimpleChannelHandler {
if (StringUtils.isNotEmpty(ack.getDestination()) && StringUtils.isNotEmpty(ack.getClientId())) {
if (ack.getBatchId() == 0L) {
NettyUtils.error(402,
MessageFormatter.format("batchId should assign value", ack.toString()).getMessage(),
ctx.getChannel(), null);
MessageFormatter.format("batchId should assign value", ack.toString()).getMessage(),
ctx.getChannel(),
null);
} else if (ack.getBatchId() == -1L) { // -1代表上一次get没有数据，直接忽略之
// donothing
} else {
@@ -155,8 +166,9 @@ public class SessionHandler extends SimpleChannelHandler {
}
} else {
NettyUtils.error(401,
MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage(),
ctx.getChannel(), null);
MessageFormatter.format("destination or clientId is null", ack.toString()).getMessage(),
ctx.getChannel(),
null);
}
break;
case CLIENTROLLBACK:
@@ -165,7 +177,7 @@ public class SessionHandler extends SimpleChannelHandler {
if (StringUtils.isNotEmpty(rollback.getDestination())
&& StringUtils.isNotEmpty(rollback.getClientId())) {
clientIdentity = new ClientIdentity(rollback.getDestination(),
Short.valueOf(rollback.getClientId()));
Short.valueOf(rollback.getClientId()));
if (rollback.getBatchId() == 0L) {
embeddedServer.rollback(clientIdentity);// 回滚所有批次
} else {
@@ -173,29 +185,33 @@ public class SessionHandler extends SimpleChannelHandler {
}
} else {
NettyUtils.error(401,
MessageFormatter.format("destination or clientId is null", rollback.toString()).getMessage(),
ctx.getChannel(), null);
MessageFormatter.format("destination or clientId is null", rollback.toString())
.getMessage(),
ctx.getChannel(),
null);
}
break;
default:
NettyUtils.error(400,
MessageFormatter.format("packet type={} is NOT supported!", packet.getType()).getMessage(),
ctx.getChannel(), null);
NettyUtils.error(400, MessageFormatter.format("packet type={} is NOT supported!", packet.getType())
.getMessage(), ctx.getChannel(), null);
break;
}
} catch (Throwable exception) {
NettyUtils.error(400,
MessageFormatter.format("something goes wrong with channel:{}, exception={}",
ctx.getChannel(), ExceptionUtils.getStackTrace(exception)).getMessage(),
ctx.getChannel(), null);
MessageFormatter.format("something goes wrong with channel:{}, exception={}",
ctx.getChannel(),
ExceptionUtils.getStackTrace(exception)).getMessage(),
ctx.getChannel(),
null);
} finally {
MDC.remove("destination");
}
}
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
logger.error("something goes wrong with channel:{}, exception={}", ctx.getChannel(),
ExceptionUtils.getStackTrace(e.getCause()));
logger.error("something goes wrong with channel:{}, exception={}",
ctx.getChannel(),
ExceptionUtils.getStackTrace(e.getCause()));
ctx.getChannel().close();
}

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

@@ -18,16 +18,16 @@ import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
public abstract class BaseCanalServerWithEmbededTest {
protected static final String cluster1 = "127.0.0.1:2188";
protected static final String DESTINATION = "ljhtest1";
protected static final String DETECTING_SQL = "insert into retl.xdual values(1,now()) on duplicate key update x=now()";
protected static final String MYSQL_ADDRESS = "127.0.0.1";
protected static final String USERNAME = "retl";
protected static final String PASSWORD = "retl";
protected static final String FILTER = "retl\\..*,erosa.zk_complaint_bizdata";
protected static final String cluster1 = "127.0.0.1:2188";
protected static final String DESTINATION = "ljhtest1";
protected static final String DETECTING_SQL = "insert into retl.xdual values(1,now()) on duplicate key update x=now()";
protected static final String MYSQL_ADDRESS = "127.0.0.1";
protected static final String USERNAME = "retl";
protected static final String PASSWORD = "retl";
protected static final String FILTER = "retl\\..*,erosa.zk_complaint_bizdata";
private CanalServerWithEmbedded server;
private ClientIdentity clientIdentity = new ClientIdentity(DESTINATION, (short) 1); ;
private ClientIdentity clientIdentity = new ClientIdentity(DESTINATION, (short) 1); ;
@Before
public void setUp() {

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

@@ -50,11 +50,9 @@ public class CanalServerWithNettyTest {
@Before
public void setUp() {
CanalServerWithEmbedded embeddedServer = new CanalServerWithEmbedded();
embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator()
{
embeddedServer.setCanalInstanceGenerator(new CanalInstanceGenerator() {
public CanalInstance generate(String destination)
{
public CanalInstance generate(String destination) {
Canal canal = buildCanal();
return new CanalInstanceWithManager(canal, FILTER);
}