diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
index f02942ae..d68d5871 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java
@@ -32,6 +32,8 @@ public class CanalConstants {
public static final String CANAL_DESTINATION_PROPERTY = ROOT + ".instance.destination";
+ public static final String CANAL_SOCKETCHANNEL = ROOT + "." + "socketChannel";
+
public static String getInstanceModeKey(String destination) {
return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination);
}
diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
index dff994cb..9bf42bf1 100644
--- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
+++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalController.java
@@ -85,6 +85,12 @@ public class CanalController {
// 初始化instance config
initInstanceConfig(properties);
+ // init socketChannel
+ String socketChannel = getProperty(properties, CanalConstants.CANAL_SOCKETCHANNEL);
+ if (StringUtils.isNotEmpty(socketChannel)) {
+ System.setProperty(CanalConstants.CANAL_SOCKETCHANNEL, socketChannel);
+ }
+
// 准备canal server
cid = Long.valueOf(getProperty(properties, CanalConstants.CANAL_ID));
ip = getProperty(properties, CanalConstants.CANAL_IP);
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java
new file mode 100644
index 00000000..c80c877d
--- /dev/null
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java
@@ -0,0 +1,136 @@
+package com.alibaba.otter.canal.parse.driver.mysql.socket;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InterruptedIOException;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketAddress;
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+
+/**
+ * 使用BIO进行dump
+ *
+ * @author chuanyi
+ */
+public class BioSocketChannel implements SocketChannel {
+
+ static final int DEFAULT_CONNECT_TIMEOUT = 10 * 1000;
+ static final int SO_TIMEOUT = 1000;
+ private Socket socket;
+ private InputStream input;
+ private OutputStream output;
+
+ BioSocketChannel(Socket socket) throws IOException{
+ this.socket = socket;
+ this.input = socket.getInputStream();
+ this.output = socket.getOutputStream();
+ }
+
+ public void write(byte[]... buf) throws IOException {
+ OutputStream output = this.output;
+ if (output != null) {
+ for (byte[] bs : buf) {
+ output.write(bs);
+ }
+ } else {
+ throw new SocketException("Socket already closed.");
+ }
+ }
+
+ public byte[] read(int readSize) throws IOException {
+ InputStream input = this.input;
+ byte[] data = new byte[readSize];
+ int remain = readSize;
+ if (input == null) {
+ throw new SocketException("Socket already closed.");
+ }
+ while (remain > 0) {
+ try {
+ int read = input.read(data, readSize - remain, remain);
+ if (read > -1) {
+ remain -= read;
+ } else {
+ throw new IOException("EOF encountered.");
+ }
+ } catch (SocketTimeoutException te) {
+ if (Thread.interrupted()) {
+ throw new InterruptedIOException("Interrupted while reading.");
+ }
+ }
+ }
+ return data;
+ }
+
+ public byte[] read(int readSize, int timeout) throws IOException {
+ InputStream input = this.input;
+ byte[] data = new byte[readSize];
+ int remain = readSize;
+ int accTimeout = 0;
+ if (input == null) {
+ throw new SocketException("Socket already closed.");
+ }
+ while (remain > 0 && accTimeout < timeout) {
+ try {
+ int read = input.read(data, readSize - remain, remain);
+ if (read > -1) {
+ remain -= read;
+ } else {
+ throw new IOException("EOF encountered.");
+ }
+ } catch (SocketTimeoutException te) {
+ if (Thread.interrupted()) {
+ throw new InterruptedIOException("Interrupted while reading.");
+ }
+ accTimeout += SO_TIMEOUT;
+ }
+ }
+ if (remain > 0 && accTimeout >= timeout) {
+ throw new SocketTimeoutException("Timeout occurred, failed to read " + readSize + " bytes in " + timeout
+ + " milliseconds.");
+ }
+ return data;
+ }
+
+ public boolean isConnected() {
+ Socket socket = this.socket;
+ if (socket != null) {
+ return socket.isConnected();
+ }
+ return false;
+ }
+
+ public SocketAddress getRemoteSocketAddress() {
+ Socket socket = this.socket;
+ if (socket != null) {
+ return socket.getRemoteSocketAddress();
+ }
+ return null;
+ }
+
+ public void close() {
+ Socket socket = this.socket;
+ if (socket != null) {
+ try {
+ socket.shutdownInput();
+ } catch (IOException e) {
+ // Ignore, could not do anymore
+ }
+ try {
+ socket.shutdownOutput();
+ } catch (IOException e) {
+ // Ignore, could not do anymore
+ }
+ try {
+ socket.close();
+ } catch (IOException e) {
+ // Ignore, could not do anymore
+ }
+ }
+ this.input = null;
+ this.output = null;
+ this.socket = null;
+ }
+
+}
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannelPool.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannelPool.java
new file mode 100644
index 00000000..3af1d469
--- /dev/null
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannelPool.java
@@ -0,0 +1,24 @@
+package com.alibaba.otter.canal.parse.driver.mysql.socket;
+
+import java.net.Socket;
+import java.net.SocketAddress;
+
+/**
+ * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
+ * @author chuanyi 2018-3-3 保留open减少文件变更数量
+ */
+public abstract class BioSocketChannelPool {
+
+ public static BioSocketChannel open(SocketAddress address) throws Exception {
+ Socket socket = new Socket();
+ socket.setReceiveBufferSize(32 * 1024);
+ socket.setSendBufferSize(32 * 1024);
+ socket.setSoTimeout(BioSocketChannel.SO_TIMEOUT);
+ socket.setTcpNoDelay(true);
+ socket.setKeepAlive(true);
+ socket.setReuseAddress(true);
+ socket.connect(address, BioSocketChannel.DEFAULT_CONNECT_TIMEOUT);
+ return new BioSocketChannel(socket);
+ }
+
+}
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java
new file mode 100644
index 00000000..47859c8e
--- /dev/null
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java
@@ -0,0 +1,143 @@
+package com.alibaba.otter.canal.parse.driver.mysql.socket;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.Channel;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+
+/**
+ * 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28
+ *
+ * @author luoyaogui
+ */
+public class NettySocketChannel implements SocketChannel {
+
+ private static final int WAIT_PERIOD = 10; // milliseconds
+ private static final int DEFAULT_INIT_BUFFER_SIZE = 1024 * 1024; // 1MB，默认初始缓存大小
+ private static final int DEFAULT_MAX_BUFFER_SIZE = 4 * DEFAULT_INIT_BUFFER_SIZE; // 4MB，默认最大缓存大小
+ private Channel channel = null;
+ private Object lock = new Object();
+ private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(DEFAULT_INIT_BUFFER_SIZE); // 缓存大小
+
+ public Channel getChannel() {
+ return channel;
+ }
+
+ public void setChannel(Channel channel) {
+ this.channel = channel;
+ }
+
+ public void writeCache(ByteBuf buf) throws InterruptedException, IOException {
+ synchronized (lock) {
+ while (true) {
+ if (null == cache) {
+ throw new IOException("socket is closed !");
+ }
+
+ cache.discardReadBytes();// 回收内存
+ // source buffer is empty.
+ if (!buf.isReadable()) {
+ break;
+ }
+
+ if (cache.isWritable()) {
+ cache.writeBytes(buf, Math.min(cache.writableBytes(), buf.readableBytes()));
+ } else {
+ // dest buffer is full.
+ lock.wait(WAIT_PERIOD);
+ }
+ }
+ }
+ }
+
+ public void write(byte[]... buf) throws IOException {
+ if (channel != null && channel.isWritable()) {
+ channel.writeAndFlush(Unpooled.copiedBuffer(buf));
+ } else {
+ throw new IOException("write failed ! please checking !");
+ }
+ }
+
+ public byte[] read(int readSize) throws IOException {
+ return read(readSize, 0);
+ }
+
+ public byte[] read(int readSize, int timeout) throws IOException {
+ int accumulatedWaitTime = 0;
+
+ // 若读取内容较长，则自动扩充超时时间，以初始缓存大小为基准计算倍数
+ if (timeout > 0 && readSize > DEFAULT_INIT_BUFFER_SIZE ) {
+ timeout *= (readSize / DEFAULT_INIT_BUFFER_SIZE + 1);
+ }
+ do {
+ if (readSize > cache.readableBytes()) {
+ if (null == channel) {
+ throw new IOException("socket has Interrupted !");
+ }
+
+ // 默认缓存大小不够用时需自动扩充，否则将因缓存空间不足而造成I/O超时假象
+ if (!cache.isWritable(readSize - cache.readableBytes())) {
+ synchronized (lock) {
+ int deltaSize = readSize - cache.readableBytes(); // 同步锁后重新读取
+ deltaSize = deltaSize - cache.writableBytes();
+ if (deltaSize > 0) {
+ deltaSize = (deltaSize / 32 + 1) * 32;
+ cache.capacity(cache.capacity() + deltaSize);
+ }
+ }
+ } else if (timeout > 0) {
+ accumulatedWaitTime += WAIT_PERIOD;
+ if (accumulatedWaitTime > timeout) {
+ StringBuilder sb = new StringBuilder("socket read timeout occured !");
+ sb.append(" readSize = ").append(readSize);
+ sb.append(", readableBytes = ").append(cache.readableBytes());
+ sb.append(", timeout = ").append(timeout);
+ throw new IOException(sb.toString());
+ }
+ }
+
+ synchronized (this) {
+ try {
+ wait(WAIT_PERIOD);
+ } catch (InterruptedException e) {
+ throw new IOException("socket has Interrupted !");
+ }
+ }
+ } else {
+ byte[] back = new byte[readSize];
+ synchronized (lock) {
+ cache.readBytes(back);
+ // 恢复自动扩充的过大缓存到默认初始大小，释放空间
+ if (cache.capacity() > DEFAULT_MAX_BUFFER_SIZE) {
+ cache.discardReadBytes(); // 回收已读空间，重置读写指针
+ if (cache.readableBytes() < DEFAULT_INIT_BUFFER_SIZE) {
+ cache.capacity(DEFAULT_INIT_BUFFER_SIZE);
+ }
+ }
+ }
+ return back;
+ }
+ } while (true);
+ }
+
+ public boolean isConnected() {
+ return channel != null ? true : false;
+ }
+
+ public SocketAddress getRemoteSocketAddress() {
+ return channel != null ? channel.remoteAddress() : null;
+ }
+
+ public void close() {
+ if (channel != null) {
+ channel.close();
+ }
+ channel = null;
+ cache.discardReadBytes();// 回收已占用的内存
+ cache.release();// 释放整个内存
+ cache = null;
+ }
+}
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannelPool.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannelPool.java
new file mode 100644
index 00000000..e97f6a4f
--- /dev/null
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannelPool.java
@@ -0,0 +1,111 @@
+package com.alibaba.otter.canal.parse.driver.mysql.socket;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import io.netty.channel.AdaptiveRecvByteBufAllocator;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+
+import java.io.IOException;
+import java.net.SocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
+ */
+@SuppressWarnings({ "rawtypes", "deprecation" })
+public abstract class NettySocketChannelPool {
+
+ private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组
+ private static Bootstrap boot = new Bootstrap(); // 主
+ private static Map chManager = new ConcurrentHashMap();
+ private static final Logger logger = LoggerFactory.getLogger(NettySocketChannelPool.class);
+
+ static {
+ boot.group(group)
+ .channel(NioSocketChannel.class)
+ .option(ChannelOption.SO_RCVBUF, 32 * 1024)
+ .option(ChannelOption.SO_SNDBUF, 32 * 1024)
+ .option(ChannelOption.TCP_NODELAY, true)
+ // 如果是延时敏感型应用，建议关闭Nagle算法
+ .option(ChannelOption.SO_KEEPALIVE, true)
+ .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
+ .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
+ //
+ .handler(new ChannelInitializer() {
+
+ @Override
+ protected void initChannel(Channel ch) throws Exception {
+ ch.pipeline().addLast(new BusinessHandler());// 命令过滤和handler添加管理
+ }
+ });
+ }
+
+ public static SocketChannel open(SocketAddress address) throws Exception {
+ SocketChannel socket = null;
+ ChannelFuture future = boot.connect(address).sync();
+
+ if (future.isSuccess()) {
+ future.channel().pipeline().get(BusinessHandler.class).latch.await();
+ socket = chManager.get(future.channel());
+ }
+
+ if (null == socket) {
+ throw new IOException("can't create socket!");
+ }
+
+ return socket;
+ }
+
+ public static class BusinessHandler extends SimpleChannelInboundHandler {
+
+ private NettySocketChannel socket = null;
+ private final CountDownLatch latch = new CountDownLatch(1);
+
+ @Override
+ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+ socket.setChannel(null);
+ chManager.remove(ctx.channel());// 移除
+ super.channelInactive(ctx);
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ socket = new NettySocketChannel();
+ socket.setChannel(ctx.channel());
+ chManager.put(ctx.channel(), socket);
+ latch.countDown();
+ super.channelActive(ctx);
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
+ if (socket != null) {
+ socket.writeCache(msg);
+ } else {
+ // TODO: need graceful error handler.
+ logger.error("no socket available.");
+ }
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ // need output error for troubeshooting.
+ logger.error("business error.", cause);
+ ctx.close();
+ }
+ }
+}
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
index 968e0eeb..952fde53 100644
--- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannel.java
@@ -1,143 +1,23 @@
package com.alibaba.otter.canal.parse.driver.mysql.socket;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.buffer.Unpooled;
-import io.netty.channel.Channel;
-
import java.io.IOException;
import java.net.SocketAddress;
/**
- * 封装netty的通信channel和数据接收缓存，实现读、写、连接校验的功能。 2016-12-28
- *
- * @author luoyaogui
+ * @author agapple 2018年3月12日 下午10:36:44
+ * @since 1.0.26
*/
-public class SocketChannel {
+public interface SocketChannel {
- private static final int WAIT_PERIOD = 10; // milliseconds
- private static final int DEFAULT_INIT_BUFFER_SIZE = 1024 * 1024; // 1MB，默认初始缓存大小
- private static final int DEFAULT_MAX_BUFFER_SIZE = 4 * DEFAULT_INIT_BUFFER_SIZE; // 4MB，默认最大缓存大小
- private Channel channel = null;
- private Object lock = new Object();
- private ByteBuf cache = PooledByteBufAllocator.DEFAULT.directBuffer(DEFAULT_INIT_BUFFER_SIZE); // 缓存大小
+ public void write(byte[]... buf) throws IOException;
- public Channel getChannel() {
- return channel;
- }
+ public byte[] read(int readSize) throws IOException;
- public void setChannel(Channel channel) {
- this.channel = channel;
- }
+ public byte[] read(int readSize, int timeout) throws IOException;
- public void writeCache(ByteBuf buf) throws InterruptedException, IOException {
- synchronized (lock) {
- while (true) {
- if (null == cache) {
- throw new IOException("socket is closed !");
- }
+ public boolean isConnected();
- cache.discardReadBytes();// 回收内存
- // source buffer is empty.
- if (!buf.isReadable()) {
- break;
- }
+ public SocketAddress getRemoteSocketAddress();
- if (cache.isWritable()) {
- cache.writeBytes(buf, Math.min(cache.writableBytes(), buf.readableBytes()));
- } else {
- // dest buffer is full.
- lock.wait(WAIT_PERIOD);
- }
- }
- }
- }
-
- public void writeChannel(byte[]... buf) throws IOException {
- if (channel != null && channel.isWritable()) {
- channel.writeAndFlush(Unpooled.copiedBuffer(buf));
- } else {
- throw new IOException("write failed ! please checking !");
- }
- }
-
- public byte[] read(int readSize) throws IOException {
- return read(readSize, 0);
- }
-
- public byte[] read(int readSize, int timeout) throws IOException {
- int accumulatedWaitTime = 0;
-
- // 若读取内容较长，则自动扩充超时时间，以初始缓存大小为基准计算倍数
- if (timeout > 0 && readSize > DEFAULT_INIT_BUFFER_SIZE ) {
- timeout *= (readSize / DEFAULT_INIT_BUFFER_SIZE + 1);
- }
- do {
- if (readSize > cache.readableBytes()) {
- if (null == channel) {
- throw new IOException("socket has Interrupted !");
- }
-
- // 默认缓存大小不够用时需自动扩充，否则将因缓存空间不足而造成I/O超时假象
- if (!cache.isWritable(readSize - cache.readableBytes())) {
- synchronized (lock) {
- int deltaSize = readSize - cache.readableBytes(); // 同步锁后重新读取
- deltaSize = deltaSize - cache.writableBytes();
- if (deltaSize > 0) {
- deltaSize = (deltaSize / 32 + 1) * 32;
- cache.capacity(cache.capacity() + deltaSize);
- }
- }
- } else if (timeout > 0) {
- accumulatedWaitTime += WAIT_PERIOD;
- if (accumulatedWaitTime > timeout) {
- StringBuilder sb = new StringBuilder("socket read timeout occured !");
- sb.append(" readSize = ").append(readSize);
- sb.append(", readableBytes = ").append(cache.readableBytes());
- sb.append(", timeout = ").append(timeout);
- throw new IOException(sb.toString());
- }
- }
-
- synchronized (this) {
- try {
- wait(WAIT_PERIOD);
- } catch (InterruptedException e) {
- throw new IOException("socket has Interrupted !");
- }
- }
- } else {
- byte[] back = new byte[readSize];
- synchronized (lock) {
- cache.readBytes(back);
- // 恢复自动扩充的过大缓存到默认初始大小，释放空间
- if (cache.capacity() > DEFAULT_MAX_BUFFER_SIZE) {
- cache.discardReadBytes(); // 回收已读空间，重置读写指针
- if (cache.readableBytes() < DEFAULT_INIT_BUFFER_SIZE) {
- cache.capacity(DEFAULT_INIT_BUFFER_SIZE);
- }
- }
- }
- return back;
- }
- } while (true);
- }
-
- public boolean isConnected() {
- return channel != null ? true : false;
- }
-
- public SocketAddress getRemoteSocketAddress() {
- return channel != null ? channel.remoteAddress() : null;
- }
-
- public void close() {
- if (channel != null) {
- channel.close();
- }
- channel = null;
- cache.discardReadBytes();// 回收已占用的内存
- cache.release();// 释放整个内存
- cache = null;
- }
+ public void close();
}
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
index efec72c6..27ad961e 100644
--- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java
@@ -1,111 +1,35 @@
-package com.alibaba.otter.canal.parse.driver.mysql.socket;
-
-import io.netty.bootstrap.Bootstrap;
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
-import io.netty.channel.AdaptiveRecvByteBufAllocator;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelHandlerContext;
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.ChannelOption;
-import io.netty.channel.EventLoopGroup;
-import io.netty.channel.SimpleChannelInboundHandler;
-import io.netty.channel.nio.NioEventLoopGroup;
-import io.netty.channel.socket.nio.NioSocketChannel;
-
-import java.io.IOException;
-import java.net.SocketAddress;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28
- */
-@SuppressWarnings({ "rawtypes", "deprecation" })
-public abstract class SocketChannelPool {
-
- private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组
- private static Bootstrap boot = new Bootstrap(); // 主
- private static Map chManager = new ConcurrentHashMap();
- private static final Logger logger = LoggerFactory.getLogger(SocketChannelPool.class);
-
- static {
- boot.group(group)
- .channel(NioSocketChannel.class)
- .option(ChannelOption.SO_RCVBUF, 32 * 1024)
- .option(ChannelOption.SO_SNDBUF, 32 * 1024)
- .option(ChannelOption.TCP_NODELAY, true)
- // 如果是延时敏感型应用，建议关闭Nagle算法
- .option(ChannelOption.SO_KEEPALIVE, true)
- .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
- .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
- //
- .handler(new ChannelInitializer() {
-
- @Override
- protected void initChannel(Channel ch) throws Exception {
- ch.pipeline().addLast(new BusinessHandler());// 命令过滤和handler添加管理
- }
- });
- }
-
- public static SocketChannel open(SocketAddress address) throws Exception {
- SocketChannel socket = null;
- ChannelFuture future = boot.connect(address).sync();
-
- if (future.isSuccess()) {
- future.channel().pipeline().get(BusinessHandler.class).latch.await();
- socket = chManager.get(future.channel());
- }
-
- if (null == socket) {
- throw new IOException("can't create socket!");
- }
-
- return socket;
- }
-
- public static class BusinessHandler extends SimpleChannelInboundHandler {
-
- private SocketChannel socket = null;
- private final CountDownLatch latch = new CountDownLatch(1);
-
- @Override
- public void channelInactive(ChannelHandlerContext ctx) throws Exception {
- socket.setChannel(null);
- chManager.remove(ctx.channel());// 移除
- super.channelInactive(ctx);
- }
-
- @Override
- public void channelActive(ChannelHandlerContext ctx) throws Exception {
- socket = new SocketChannel();
- socket.setChannel(ctx.channel());
- chManager.put(ctx.channel(), socket);
- latch.countDown();
- super.channelActive(ctx);
- }
-
- @Override
- protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
- if (socket != null) {
- socket.writeCache(msg);
- } else {
- // TODO: need graceful error handler.
- logger.error("no socket available.");
- }
- }
-
- @Override
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
- // need output error for troubeshooting.
- logger.error("business error.", cause);
- ctx.close();
- }
- }
-}
+package com.alibaba.otter.canal.parse.driver.mysql.socket;
+
+import java.net.SocketAddress;
+
+import org.apache.commons.lang.StringUtils;
+
+/**
+ * @author agapple 2018年3月12日 下午10:46:22
+ * @since 1.0.26
+ */
+public abstract class SocketChannelPool {
+
+ public static SocketChannel open(SocketAddress address) throws Exception {
+ String type = chooseSocketChannel();
+ if ("netty".equalsIgnoreCase(type)) {
+ return NettySocketChannelPool.open(address);
+ } else {
+ return BioSocketChannelPool.open(address);
+ }
+
+ }
+
+ private static String chooseSocketChannel() {
+ String socketChannel = System.getenv("canal.socketChannel");
+ if (StringUtils.isEmpty(socketChannel)) {
+ socketChannel = System.getProperty("canal.socketChannel");
+ }
+
+ if (StringUtils.isEmpty(socketChannel)) {
+ socketChannel = "bio"; // bio or netty
+ }
+
+ return socketChannel;
+ }
+}
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java
index 7d4b9020..bdf01196 100644
--- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/PacketManager.java
@@ -28,7 +28,7 @@ public abstract class PacketManager {
}
public static void writePkg(SocketChannel ch, byte[]... srcs) throws IOException {
- ch.writeChannel(srcs);
+ ch.write(srcs);
}
public static void writeBody(SocketChannel ch, byte[] body) throws IOException {
@@ -39,6 +39,6 @@ public abstract class PacketManager {
HeaderPacket header = new HeaderPacket();
header.setPacketBodyLength(body.length);
header.setPacketSequenceNumber(packetSeqNumber);
- ch.writeChannel(header.toBytes(), body);
+ ch.write(header.toBytes(), body);
}
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
index 0eeef9f5..464c43f0 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java
@@ -1,10 +1,11 @@
package com.alibaba.otter.canal.parse.inbound.mysql;
+import static com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher.MASTER_HEARTBEAT_PERIOD_SECONDS;
+
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.List;
-import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
@@ -29,8 +30,6 @@ import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
-import static com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher.MASTER_HEARTBEAT_PERIOD_SECONDS;
-
public class MysqlConnection implements ErosaConnection {
private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class);
@@ -298,10 +297,12 @@ public class MysqlConnection implements ErosaConnection {
}
/**
- * MASTER_HEARTBEAT_PERIOD sets the interval in seconds between replication heartbeats.
- * Whenever the master's binary log is updated with an event, the waiting period for the next heartbeat is reset.
- * interval is a decimal value having the range 0 to 4294967 seconds and a resolution in milliseconds;
- * the smallest nonzero value is 0.001. Heartbeats are sent by the master only if there are no unsent events
+ * MASTER_HEARTBEAT_PERIOD sets the interval in seconds between
+ * replication heartbeats. Whenever the master's binary log is updated
+ * with an event, the waiting period for the next heartbeat is reset.
+ * interval is a decimal value having the range 0 to 4294967 seconds and
+ * a resolution in milliseconds; the smallest nonzero value is 0.001.
+ * Heartbeats are sent by the master only if there are no unsent events
* in the binary log file for a period longer than interval.
*/
try {
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
index d021868c..18803657 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java
@@ -23,8 +23,8 @@ public class DirectLogFetcher extends LogFetcher {
// Master heartbeat interval
public static final int MASTER_HEARTBEAT_PERIOD_SECONDS = 15;
- // +1s 确保 timeout > heartbeat interval
- private static final int READ_TIMEOUT_MILLISECONDS = (MASTER_HEARTBEAT_PERIOD_SECONDS + 1) * 1000;
+ // +10s 确保 timeout > heartbeat interval
+ private static final int READ_TIMEOUT_MILLISECONDS = (MASTER_HEARTBEAT_PERIOD_SECONDS + 10) * 1000;
/** Command to dump binlog */
public static final byte COM_BINLOG_DUMP = 18;
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java
index 54ea60a8..4c1cef3c 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java
@@ -2,15 +2,19 @@ package com.alibaba.otter.canal.parse.inbound.mysql.dbsync;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.commons.lang.StringUtils;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.TableMeta;
import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
import com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection;
+import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.DatabaseTableMeta;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.MemoryTableMeta;
import com.alibaba.otter.canal.parse.inbound.mysql.tsdb.TableMetaTSDB;
@@ -75,14 +79,22 @@ public class TableMetaCache {
}
private TableMeta getTableMetaByDB(String fullname) throws IOException {
- ResultSetPacket packet = connection.query("show create table " + fullname);
- String[] names = StringUtils.split(fullname, "`.`");
- String schema = names[0];
- String table = names[1].substring(0, names[1].length());
- return new TableMeta(schema, table, parserTableMeta(schema, table, packet));
+ try {
+ ResultSetPacket packet = connection.query("show create table " + fullname);
+ String[] names = StringUtils.split(fullname, "`.`");
+ String schema = names[0];
+ String table = names[1].substring(0, names[1].length());
+ return new TableMeta(schema, table, parseTableMeta(schema, table, packet));
+ } catch (Throwable e) { // fallback to desc table
+ ResultSetPacket packet = connection.query("desc " + fullname);
+ String[] names = StringUtils.split(fullname, "`.`");
+ String schema = names[0];
+ String table = names[1].substring(0, names[1].length());
+ return new TableMeta(schema, table, parseTableMetaByDesc(packet));
+ }
}
- public static List parserTableMeta(String schema, String table, ResultSetPacket packet) {
+ public static List parseTableMeta(String schema, String table, ResultSetPacket packet) {
if (packet.getFieldValues().size() > 1) {
String createDDL = packet.getFieldValues().get(1);
MemoryTableMeta memoryTableMeta = new MemoryTableMeta();
@@ -94,6 +106,40 @@ public class TableMetaCache {
}
}
+ /**
+ * 处理desc table的结果
+ */
+ public static List parseTableMetaByDesc(ResultSetPacket packet) {
+ Map nameMaps = new HashMap(6, 1f);
+ int index = 0;
+ for (FieldPacket fieldPacket : packet.getFieldDescriptors()) {
+ nameMaps.put(fieldPacket.getOriginalName(), index++);
+ }
+
+ int size = packet.getFieldDescriptors().size();
+ int count = packet.getFieldValues().size() / packet.getFieldDescriptors().size();
+ List result = new ArrayList();
+ for (int i = 0; i < count; i++) {
+ FieldMeta meta = new FieldMeta();
+ // 做一个优化，使用String.intern()，共享String对象，减少内存使用
+ meta.setColumnName(packet.getFieldValues().get(nameMaps.get(COLUMN_NAME) + i * size).intern());
+ meta.setColumnType(packet.getFieldValues().get(nameMaps.get(COLUMN_TYPE) + i * size));
+ meta.setNullable(StringUtils.equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(IS_NULLABLE) + i
+ * size),
+ "YES"));
+ meta.setKey("PRI".equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(COLUMN_KEY) + i * size)));
+ meta.setUnique("UNI".equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(COLUMN_KEY) + i * size)));
+ // 特殊处理引号
+ meta.setDefaultValue(DruidDdlParser.unescapeQuotaName(packet.getFieldValues()
+ .get(nameMaps.get(COLUMN_DEFAULT) + i * size)));
+ meta.setExtra(packet.getFieldValues().get(nameMaps.get(EXTRA) + i * size));
+
+ result.add(meta);
+ }
+
+ return result;
+ }
+
public TableMeta getTableMeta(String schema, String table) {
return getTableMeta(schema, table, true);
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
index 550d8a00..4b917f2f 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
@@ -296,7 +296,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
ResultSetPacket packet = connection.query("show create table " + getFullName(schema, table));
if (packet.getFieldValues().size() > 1) {
createDDL = packet.getFieldValues().get(1);
- tableMetaFromDB.setFields(TableMetaCache.parserTableMeta(schema, table, packet));
+ tableMetaFromDB.setFields(TableMetaCache.parseTableMeta(schema, table, packet));
}
} catch (IOException e) {
if (e.getMessage().contains("errorNumber=1146")) {