From 6b7109c66b19f5bf97ad7473e0131c50ea8924f7 Mon Sep 17 00:00:00 2001 From: agapple Date: Mon, 12 Mar 2018 23:21:43 +0800 Subject: [PATCH] fixed issue #539 , support bio&netty --- .../otter/canal/deployer/CanalConstants.java | 2 + .../otter/canal/deployer/CanalController.java | 6 + .../driver/mysql/socket/BioSocketChannel.java | 136 ++++++++++++++++ .../mysql/socket/BioSocketChannelPool.java | 24 +++ .../mysql/socket/NettySocketChannel.java | 143 +++++++++++++++++ .../mysql/socket/NettySocketChannelPool.java | 111 +++++++++++++ .../driver/mysql/socket/SocketChannel.java | 138 ++--------------- .../mysql/socket/SocketChannelPool.java | 146 +++++------------- .../driver/mysql/utils/PacketManager.java | 4 +- .../parse/inbound/mysql/MysqlConnection.java | 15 +- .../mysql/dbsync/DirectLogFetcher.java | 4 +- .../inbound/mysql/dbsync/TableMetaCache.java | 58 ++++++- .../inbound/mysql/tsdb/DatabaseTableMeta.java | 2 +- 13 files changed, 531 insertions(+), 258 deletions(-) create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannelPool.java create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannel.java create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/NettySocketChannelPool.java 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")) {