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

format & import

Browse Source
This commit is contained in:
七锋
2018-09-30 10:52:41 +08:00
parent c3327e567e
commit 9344ed121f
46 changed files with 458 additions and 464 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/CanalOuterAdapter.java
View File

@@ -3,8 +3,6 @@ package com.alibaba.otter.canal.client.adapter;
import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.SPI; import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
/** /**
* 外部适配器接口 * 外部适配器接口

2
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java
View File

@@ -24,7 +24,6 @@ public class CanalClientConfig {
private Boolean flatMessage = true; private Boolean flatMessage = true;
private List<CanalInstance> canalInstances; private List<CanalInstance> canalInstances;
public String getCanalServerHost() { public String getCanalServerHost() {
@@ -122,6 +121,7 @@ public class CanalClientConfig {
} }
public static class MQTopic { public static class MQTopic {
private String mqMode; private String mqMode;
private String topic; private String topic;

6
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalOuterAdapterConfiguration.java
View File

@@ -10,11 +10,11 @@ import java.util.Properties;
*/ */
public class CanalOuterAdapterConfiguration { public class CanalOuterAdapterConfiguration {
private String name; // 适配器名称, 如: logger, hbase, es private String name; // 适配器名称, 如: logger, hbase, es
private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址 private String hosts; // 适配器内部的地址, 比如对应es该参数可以填写es的server地址
private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址 private String zkHosts; // 适配器内部的ZK地址, 比如对应HBase该参数可以填写HBase对应的ZK地址
private Properties properties; // 其余参数, 可填写适配器中的所需的配置信息 private Properties properties; // 其余参数, 可填写适配器中的所需的配置信息

4
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/Dml.java
View File

@@ -90,7 +90,7 @@ public class Dml implements Serializable {
@Override @Override
public String toString() { public String toString() {
return "Dml{" + "database='" + database + '\'' + ", table='" + table + '\'' + ", type='" + type + '\'' + ", ts=" return "Dml{" + "database='" + database + '\'' + ", table='" + table + '\'' + ", type='" + type + '\''
+ ts + ", sql='" + sql + '\'' + ", data=" + data + ", old=" + old + '}'; + ", ts=" + ts + ", sql='" + sql + '\'' + ", data=" + data + ", old=" + old + '}';
} }
} }

90
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java
View File

@@ -1,11 +1,22 @@
package com.alibaba.otter.canal.client.adapter.support; package com.alibaba.otter.canal.client.adapter.support;
import java.io.*; import java.io.BufferedReader;
import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.MalformedURLException; import java.net.MalformedURLException;
import java.net.URL; import java.net.URL;
import java.net.URLClassLoader; import java.net.URLClassLoader;
import java.nio.file.Paths; import java.nio.file.Paths;
import java.util.*; import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.regex.Pattern; import java.util.regex.Pattern;
@@ -21,8 +32,7 @@ import org.slf4j.LoggerFactory;
*/ */
public class ExtensionLoader<T> { public class ExtensionLoader<T> {
private static final Logger logger = LoggerFactory private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class);
.getLogger(ExtensionLoader.class);
private static final String SERVICES_DIRECTORY = "META-INF/services/"; private static final String SERVICES_DIRECTORY = "META-INF/services/";
@@ -30,8 +40,7 @@ public class ExtensionLoader<T> {
private static final String DEFAULT_CLASSLOADER_POLICY = "internal"; private static final String DEFAULT_CLASSLOADER_POLICY = "internal";
private static final Pattern NAME_SEPARATOR = Pattern private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*");
.compile("\\s*[,]+\\s*");
private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>(); private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>();
@@ -262,8 +271,7 @@ public class ExtensionLoader<T> {
return instance; return instance;
} catch (Throwable t) { } catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+ ") could not be instantiated: " + t.getMessage(), + ") could not be instantiated: " + t.getMessage(), t);
t);
} }
} }
@@ -271,8 +279,8 @@ public class ExtensionLoader<T> {
if (type == null) throw new IllegalArgumentException("Extension type == null"); if (type == null) throw new IllegalArgumentException("Extension type == null");
if (name == null) throw new IllegalArgumentException("Extension name == null"); if (name == null) throw new IllegalArgumentException("Extension name == null");
Class<?> clazz = getExtensionClasses().get(name); Class<?> clazz = getExtensionClasses().get(name);
if (clazz == null) if (clazz == null) throw new IllegalStateException("No such extension \"" + name + "\" for " + type.getName()
throw new IllegalStateException("No such extension \"" + name + "\" for " + type.getName() + "!"); + "!");
return clazz; return clazz;
} }
@@ -296,10 +304,17 @@ public class ExtensionLoader<T> {
throw new IllegalStateException("failed to get class loader resource"); throw new IllegalStateException("failed to get class loader resource");
} }
String dirtyPath = url.toString(); String dirtyPath = url.toString();
String jarPath = dirtyPath.replaceAll("^.*file:/", ""); // removes file:/ and everything before it String jarPath = dirtyPath.replaceAll("^.*file:/", ""); // removes
jarPath = jarPath.replaceAll("jar!.*", "jar"); // removes everything after .jar, if .jar exists in dirtyPath // file:/ and
jarPath = jarPath.replaceAll("%20", " "); // necessary if path has spaces within // everything
if (!jarPath.endsWith(".jar")) { // this is needed if you plan to run the app using Spring Tools Suit play // before it
jarPath = jarPath.replaceAll("jar!.*", "jar"); // removes everything
// after .jar, if .jar
// exists in dirtyPath
jarPath = jarPath.replaceAll("%20", " "); // necessary if path has
// spaces within
if (!jarPath.endsWith(".jar")) { // this is needed if you plan to run
// the app using Spring Tools Suit play
// button. // button.
jarPath = jarPath.replaceAll("/classes/.*", "/classes/"); jarPath = jarPath.replaceAll("/classes/.*", "/classes/");
} }
@@ -327,8 +342,8 @@ public class ExtensionLoader<T> {
logger.info("extension classpath dir: " + dir); logger.info("extension classpath dir: " + dir);
File externalLibDir = new File(dir); File externalLibDir = new File(dir);
if (!externalLibDir.exists()) { if (!externalLibDir.exists()) {
externalLibDir = new File( externalLibDir = new File(File.separator + this.getJarDirectoryPath() + File.separator + "canal_client"
File.separator + this.getJarDirectoryPath() + File.separator + "canal_client" + File.separator + "lib"); + File.separator + "lib");
} }
if (externalLibDir.exists()) { if (externalLibDir.exists()) {
File[] files = externalLibDir.listFiles(new FilenameFilter() { File[] files = externalLibDir.listFiles(new FilenameFilter() {
@@ -363,7 +378,8 @@ public class ExtensionLoader<T> {
if (name.startsWith("java.") || name.startsWith("org.slf4j.") if (name.startsWith("java.") || name.startsWith("org.slf4j.")
|| name.startsWith("org.apache.logging") || name.startsWith("org.apache.logging")
|| name.startsWith("org.apache.commons.logging.")) { || name.startsWith("org.apache.commons.logging.")) {
// || name.startsWith("org.apache.hadoop.")) { // || name.startsWith("org.apache.hadoop."))
// {
c = super.loadClass(name); c = super.loadClass(name);
} }
if (c != null) return c; if (c != null) return c;
@@ -386,7 +402,8 @@ public class ExtensionLoader<T> {
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
Enumeration<URL>[] tmp = (Enumeration<URL>[]) new Enumeration<?>[2]; Enumeration<URL>[] tmp = (Enumeration<URL>[]) new Enumeration<?>[2];
tmp[0] = findResources(name); // local class path first tmp[0] = findResources(name); // local class
// path first
// tmp[1] = super.getResources(name); // tmp[1] = super.getResources(name);
return new CompoundEnumeration<>(tmp); return new CompoundEnumeration<>(tmp);
@@ -474,12 +491,16 @@ public class ExtensionLoader<T> {
} }
if (line.length() > 0) { if (line.length() > 0) {
Class<?> clazz = classLoader.loadClass(line); Class<?> clazz = classLoader.loadClass(line);
// Class<?> clazz = Class.forName(line, true, classLoader); // Class<?> clazz =
// Class.forName(line, true,
// classLoader);
if (!type.isAssignableFrom(clazz)) { if (!type.isAssignableFrom(clazz)) {
throw new IllegalStateException( throw new IllegalStateException("Error when load extension class(interface: "
"Error when load extension class(interface: " + type + type
+ ", class line: " + clazz.getName() + ", class line: "
+ "), class " + clazz.getName() + clazz.getName()
+ "), class "
+ clazz.getName()
+ "is not subtype of interface."); + "is not subtype of interface.");
} else { } else {
try { try {
@@ -497,9 +518,9 @@ public class ExtensionLoader<T> {
extensionClasses.put(n, clazz); extensionClasses.put(n, clazz);
} else if (c != clazz) { } else if (c != clazz) {
cachedNames.remove(clazz); cachedNames.remove(clazz);
throw new IllegalStateException( throw new IllegalStateException("Duplicate extension "
"Duplicate extension " + type.getName() + " name " + type.getName()
+ n + " on " + " name " + n + " on "
+ c.getName() + " and " + c.getName() + " and "
+ clazz.getName()); + clazz.getName());
} }
@@ -509,9 +530,12 @@ public class ExtensionLoader<T> {
} }
} }
} catch (Throwable t) { } catch (Throwable t) {
IllegalStateException e = new IllegalStateException( IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: "
"Failed to load extension class(interface: " + type + ", class line: " + type
+ line + ") in " + url + ", class line: "
+ line
+ ") in "
+ url
+ ", cause: " + ", cause: "
+ t.getMessage(), + t.getMessage(),
t); t);
@@ -526,15 +550,13 @@ public class ExtensionLoader<T> {
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + type + ", class file: " + url logger.error("Exception when load extension class(interface: " + type + ", class file: " + url
+ ") in " + url, + ") in " + url, t);
t);
} }
} // end of while urls } // end of while urls
} }
} catch (Throwable t) { } catch (Throwable t) {
logger.error( logger.error("Exception when load extension class(interface: " + type + ", description file: " + fileName
"Exception when load extension class(interface: " + type + ", description file: " + fileName + ").", + ").", t);
t);
} }
} }

16
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/MessageUtil.java
View File

@@ -1,6 +1,11 @@
package com.alibaba.otter.canal.client.adapter.support; package com.alibaba.otter.canal.client.adapter.support;
import java.util.*; import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.FlatMessage;
@@ -81,11 +86,10 @@ public class MessageUtil {
Map<String, Object> rowOld = new LinkedHashMap<>(); Map<String, Object> rowOld = new LinkedHashMap<>();
for (CanalEntry.Column column : rowData.getBeforeColumnsList()) { for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
if (updateSet.contains(column.getName())) { if (updateSet.contains(column.getName())) {
rowOld.put(column.getName(), rowOld.put(column.getName(), JdbcTypeUtil.typeConvert(column.getName(),
JdbcTypeUtil.typeConvert(column.getName(), column.getValue(),
column.getValue(), column.getSqlType(),
column.getSqlType(), column.getMysqlType()));
column.getMysqlType()));
} }
} }
// update操作将记录修改前的值 // update操作将记录修改前的值

6
client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/SPI.java
View File

@@ -1,6 +1,10 @@
package com.alibaba.otter.canal.client.adapter.support; package com.alibaba.otter.canal.client.adapter.support;
import java.lang.annotation.*; import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
@Documented @Documented
@Retention(RetentionPolicy.RUNTIME) @Retention(RetentionPolicy.RUNTIME)

6
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
View File

@@ -4,7 +4,6 @@ import java.io.IOException;
import java.util.HashMap; import java.util.HashMap;
import java.util.Map; import java.util.Map;
import com.alibaba.otter.canal.protocol.FlatMessage;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.HBaseConfiguration;
@@ -17,9 +16,7 @@ import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfigLoader;
import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService; import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService;
import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration; import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
import com.alibaba.otter.canal.client.adapter.support.SPI; import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.protocol.Message;
/** /**
* HBase外部适配器 * HBase外部适配器
@@ -45,8 +42,7 @@ public class HbaseAdapter implements CanalOuterAdapter {
mappingConfigCache = new HashMap<>(); mappingConfigCache = new HashMap<>();
for (MappingConfig mappingConfig : hbaseMapping.values()) { for (MappingConfig mappingConfig : hbaseMapping.values()) {
mappingConfigCache.put(mappingConfig.getHbaseOrm().getDatabase() + "-" mappingConfigCache.put(mappingConfig.getHbaseOrm().getDatabase() + "-"
+ mappingConfig.getHbaseOrm().getTable(), + mappingConfig.getHbaseOrm().getTable(), mappingConfig);
mappingConfig);
} }
} }
} }

8
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java
View File

@@ -1,6 +1,10 @@
package com.alibaba.otter.canal.client.adapter.hbase.config; package com.alibaba.otter.canal.client.adapter.hbase.config;
import java.util.*; import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
/** /**
* HBase表映射配置 * HBase表映射配置
@@ -117,7 +121,7 @@ public class MappingConfig {
} }
public enum Mode { public enum Mode {
STRING("STRING"), NATIVE("NATIVE"), PHOENIX("PHOENIX"); STRING("STRING"), NATIVE("NATIVE"), PHOENIX("PHOENIX");
private String type; private String type;

25
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
View File

@@ -1,6 +1,10 @@
package com.alibaba.otter.canal.client.adapter.hbase.service; package com.alibaba.otter.canal.client.adapter.hbase.service;
import java.util.*; import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Bytes;
@@ -8,7 +12,12 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig; import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
import com.alibaba.otter.canal.client.adapter.hbase.support.*; import com.alibaba.otter.canal.client.adapter.hbase.support.HRow;
import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
import com.alibaba.otter.canal.client.adapter.hbase.support.PhType;
import com.alibaba.otter.canal.client.adapter.hbase.support.PhTypeUtil;
import com.alibaba.otter.canal.client.adapter.hbase.support.Type;
import com.alibaba.otter.canal.client.adapter.hbase.support.TypeUtil;
import com.alibaba.otter.canal.client.adapter.support.Dml; import com.alibaba.otter.canal.client.adapter.support.Dml;
/** /**
@@ -63,7 +72,8 @@ public class HbaseSyncService {
MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm(); MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
// if (!validHTable(config)) { // if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable()); // logger.error("HBase table '{}' not exists",
// hbaseOrm.getHbaseTable());
// return; // return;
// } // }
int i = 1; int i = 1;
@@ -156,7 +166,8 @@ public class HbaseSyncService {
MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm(); MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
// if (!validHTable(config)) { // if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable()); // logger.error("HBase table '{}' not exists",
// hbaseOrm.getHbaseTable());
// return; // return;
// } // }
@@ -248,7 +259,8 @@ public class HbaseSyncService {
MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm(); MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
// if (!validHTable(config)) { // if (!validHTable(config)) {
// logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable()); // logger.error("HBase table '{}' not exists",
// hbaseOrm.getHbaseTable());
// return; // return;
// } // }
@@ -363,8 +375,7 @@ public class HbaseSyncService {
* @param value 值 * @param value 值
* @return 复合字段rowKey * @return 复合字段rowKey
*/ */
private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseOrm hbaseOrm, private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseOrm hbaseOrm, Object value) {
Object value) {
if (value == null) { if (value == null) {
return null; return null;
} }

6
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java
View File

@@ -8,7 +8,11 @@ import java.util.Set;
import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*; import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes; import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;

36
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhType.java
View File

@@ -13,37 +13,21 @@ import java.util.Date;
* @version 1.0.0 * @version 1.0.0
*/ */
public enum PhType { public enum PhType {
DEFAULT(-1, "VARCHAR"), DEFAULT(-1, "VARCHAR"), UNSIGNED_INT(4, "UNSIGNED_INT"), UNSIGNED_LONG(8, "UNSIGNED_LONG"),
UNSIGNED_INT(4, "UNSIGNED_INT"), UNSIGNED_TINYINT(1, "UNSIGNED_TINYINT"), UNSIGNED_SMALLINT(2, "UNSIGNED_SMALLINT"),
UNSIGNED_LONG(8, "UNSIGNED_LONG"), UNSIGNED_FLOAT(4, "UNSIGNED_FLOAT"), UNSIGNED_DOUBLE(8, "UNSIGNED_DOUBLE"), INTEGER(4, "INTEGER"),
UNSIGNED_TINYINT(1, "UNSIGNED_TINYINT"), BIGINT(8, "BIGINT"), TINYINT(1, "TINYINT"), SMALLINT(2, "SMALLINT"), FLOAT(4, "FLOAT"), DOUBLE(8, "DOUBLE"),
UNSIGNED_SMALLINT(2, "UNSIGNED_SMALLINT"), DECIMAL(-1, "DECIMAL"), BOOLEAN(1, "BOOLEAN"), UNSIGNED_TIME(8, "UNSIGNED_TIME"),
UNSIGNED_FLOAT(4, "UNSIGNED_FLOAT"), UNSIGNED_DATE(8, "UNSIGNED_DATE"), UNSIGNED_TIMESTAMP(12, "UNSIGNED_TIMESTAMP"), TIME(8, "TIME"), DATE(8, "DATE"),
UNSIGNED_DOUBLE(8, "UNSIGNED_DOUBLE"), TIMESTAMP(12, "TIMESTAMP"), VARCHAR(-1, "VARCHAR"), VARBINARY(-1, "VARBINARY");
INTEGER(4, "INTEGER"),
BIGINT(8, "BIGINT"),
TINYINT(1, "TINYINT"),
SMALLINT(2, "SMALLINT"),
FLOAT(4, "FLOAT"),
DOUBLE(8, "DOUBLE"),
DECIMAL(-1, "DECIMAL"),
BOOLEAN(1, "BOOLEAN"),
UNSIGNED_TIME(8, "UNSIGNED_TIME"),
UNSIGNED_DATE(8, "UNSIGNED_DATE"),
UNSIGNED_TIMESTAMP(12, "UNSIGNED_TIMESTAMP"),
TIME(8, "TIME"),
DATE(8, "DATE"),
TIMESTAMP(12, "TIMESTAMP"),
VARCHAR(-1, "VARCHAR"),
VARBINARY(-1, "VARBINARY");
/** /**
* -1：长度可变 * -1：长度可变
*/ */
private int len; private int len;
private String type; private String type;
PhType(int len, String type) { PhType(int len, String type){
this.len = len; this.len = len;
this.type = type; this.type = type;
} }
@@ -87,7 +71,7 @@ public enum PhType {
phType = VARCHAR; phType = VARCHAR;
} else if (BigDecimal.class.isAssignableFrom(javaType)) { } else if (BigDecimal.class.isAssignableFrom(javaType)) {
phType = DECIMAL; phType = DECIMAL;
} else if (BigInteger.class.isAssignableFrom(javaType)) { } else if (BigInteger.class.isAssignableFrom(javaType)) {
phType = UNSIGNED_LONG; phType = UNSIGNED_LONG;
} else { } else {
phType = DEFAULT; phType = DEFAULT;

22
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java
View File

@@ -169,7 +169,8 @@ public class PhTypeUtil {
private static int encodeInt(int v, byte[] b, int o) { private static int encodeInt(int v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_INT); checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
b[o + 0] = (byte) ((v >> 24) ^ 0x80); // Flip sign bit so that INTEGER is binary comparable b[o + 0] = (byte) ((v >> 24) ^ 0x80); // Flip sign bit so that INTEGER
// is binary comparable
b[o + 1] = (byte) (v >> 16); b[o + 1] = (byte) (v >> 16);
b[o + 2] = (byte) (v >> 8); b[o + 2] = (byte) (v >> 8);
b[o + 3] = (byte) v; b[o + 3] = (byte) v;
@@ -209,7 +210,8 @@ public class PhTypeUtil {
private static int encodeLong(long v, byte[] b, int o) { private static int encodeLong(long v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_LONG); checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
b[o + 0] = (byte) ((v >> 56) ^ 0x80); // Flip sign bit so that INTEGER is binary comparable b[o + 0] = (byte) ((v >> 56) ^ 0x80); // Flip sign bit so that INTEGER
// is binary comparable
b[o + 1] = (byte) (v >> 48); b[o + 1] = (byte) (v >> 48);
b[o + 2] = (byte) (v >> 40); b[o + 2] = (byte) (v >> 40);
b[o + 3] = (byte) (v >> 32); b[o + 3] = (byte) (v >> 32);
@@ -255,7 +257,8 @@ public class PhTypeUtil {
private static int encodeShort(short v, byte[] b, int o) { private static int encodeShort(short v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT); checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
b[o + 0] = (byte) ((v >> 8) ^ 0x80); // Flip sign bit so that Short is binary comparable b[o + 0] = (byte) ((v >> 8) ^ 0x80); // Flip sign bit so that Short is
// binary comparable
b[o + 1] = (byte) v; b[o + 1] = (byte) v;
return Bytes.SIZEOF_SHORT; return Bytes.SIZEOF_SHORT;
} }
@@ -287,7 +290,8 @@ public class PhTypeUtil {
private static int encodeByte(byte v, byte[] b, int o) { private static int encodeByte(byte v, byte[] b, int o) {
checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE); checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE);
b[o] = (byte) (v ^ 0x80); // Flip sign bit so that Short is binary comparable b[o] = (byte) (v ^ 0x80); // Flip sign bit so that Short is binary
// comparable
return Bytes.SIZEOF_BYTE; return Bytes.SIZEOF_BYTE;
} }
@@ -546,7 +550,8 @@ public class PhTypeUtil {
multiplyBy = 10; multiplyBy = 10;
divideBy = BigInteger.TEN; divideBy = BigInteger.TEN;
} }
// Normalize the scale based on what is necessary to end up with a base 100 // Normalize the scale based on what is necessary to end up with a base
// 100
// decimal (i.e. 10.123e3) // decimal (i.e. 10.123e3)
int digitOffset; int digitOffset;
BigInteger compareAgainst; BigInteger compareAgainst;
@@ -564,7 +569,8 @@ public class PhTypeUtil {
if (length <= MAX_BIG_DECIMAL_BYTES) { if (length <= MAX_BIG_DECIMAL_BYTES) {
result[--index] = NEG_TERMINAL_BYTE; result[--index] = NEG_TERMINAL_BYTE;
} else { } else {
// Adjust length and offset down because we don't have enough room // Adjust length and offset down because we don't have enough
// room
length = MAX_BIG_DECIMAL_BYTES; length = MAX_BIG_DECIMAL_BYTES;
index = offset + length; index = offset + length;
} }
@@ -593,8 +599,8 @@ public class PhTypeUtil {
private static void checkForSufficientLength(byte[] b, int offset, int requiredLength) { private static void checkForSufficientLength(byte[] b, int offset, int requiredLength) {
if (b.length < offset + requiredLength) { if (b.length < offset + requiredLength) {
throw new RuntimeException( throw new RuntimeException("Expected length of at least " + requiredLength + " bytes, but had "
"Expected length of at least " + requiredLength + " bytes, but had " + (b.length - offset)); + (b.length - offset));
} }
} }

16
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/Type.java
View File

@@ -11,22 +11,12 @@ import java.util.Date;
* @version 1.0.0 * @version 1.0.0
*/ */
public enum Type { public enum Type {
DEFAULT("STRING"), DEFAULT("STRING"), STRING("STRING"), INTEGER("INTEGER"), LONG("LONG"), SHORT("SHORT"), BOOLEAN("BOOLEAN"),
STRING("STRING"), FLOAT("FLOAT"), DOUBLE("DOUBLE"), BIGDECIMAL("BIGDECIMAL"), DATE("DATE"), BYTE("BYTE"), BYTES("BYTES");
INTEGER("INTEGER"),
LONG("LONG"),
SHORT("SHORT"),
BOOLEAN("BOOLEAN"),
FLOAT("FLOAT"),
DOUBLE("DOUBLE"),
BIGDECIMAL("BIGDECIMAL"),
DATE("DATE"),
BYTE("BYTE"),
BYTES("BYTES");
private String type; private String type;
Type(String type) { Type(String type){
this.type = type; this.type = type;
} }

3
client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java
View File

@@ -14,7 +14,8 @@ import com.alibaba.otter.canal.client.adapter.support.SPI;
* @author machengyuan 2018-8-19 下午11:45:38 * @author machengyuan 2018-8-19 下午11:45:38
* @version 1.0.0 * @version 1.0.0
*/ */
@SPI("logger") // logger参数对应CanalOuterAdapterConfiguration配置中的name @SPI("logger")
// logger参数对应CanalOuterAdapterConfiguration配置中的name
public class LoggerAdapterExample implements CanalOuterAdapter { public class LoggerAdapterExample implements CanalOuterAdapter {
private Logger logger = LoggerFactory.getLogger(this.getClass()); private Logger logger = LoggerFactory.getLogger(this.getClass());

7
client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java
View File

@@ -1,13 +1,14 @@
package com.alibaba.otter.canal.client; package com.alibaba.otter.canal.client;
import com.alibaba.otter.canal.client.adapter.loader.CanalAdapterLoader; import java.io.FileInputStream;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml; import org.yaml.snakeyaml.Yaml;
import java.io.FileInputStream; import com.alibaba.otter.canal.client.adapter.loader.CanalAdapterLoader;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
public class ClientLauncher { public class ClientLauncher {

28
client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java
View File

@@ -26,10 +26,10 @@ public abstract class AbstractCanalAdapterWorker {
protected final Logger logger = LoggerFactory.getLogger(this.getClass()); protected final Logger logger = LoggerFactory.getLogger(this.getClass());
protected String canalDestination; // canal实例 protected String canalDestination; // canal实例
protected List<List<CanalOuterAdapter>> canalOuterAdapters; // 外部适配器 protected List<List<CanalOuterAdapter>> canalOuterAdapters; // 外部适配器
protected ExecutorService groupInnerExecutorService; // 组内工作线程池 protected ExecutorService groupInnerExecutorService; // 组内工作线程池
protected volatile boolean running = false; // 是否运行中 protected volatile boolean running = false; // 是否运行中
protected Thread thread = null; protected Thread thread = null;
protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
@@ -129,31 +129,22 @@ public abstract class AbstractCanalAdapterWorker {
} }
} }
protected void writeOut(Message message,String topic){ protected void writeOut(Message message, String topic) {
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("topic: {} batchId: {} batchSize: {} ", logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size());
topic,
message.getId(),
message.getEntries().size());
} }
long begin = System.currentTimeMillis(); long begin = System.currentTimeMillis();
writeOut(message); writeOut(message);
long now = System.currentTimeMillis(); long now = System.currentTimeMillis();
if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) { if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
logger.error("topic: {} batchId {} elapsed time: {} ms", logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
topic,
message.getId(),
now - begin);
} }
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("topic: {} batchId {} elapsed time: {} ms", logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
topic,
message.getId(),
now - begin);
} }
} }
protected void stopOutAdapters(){ protected void stopOutAdapters() {
if (thread != null) { if (thread != null) {
try { try {
thread.join(); thread.join();
@@ -170,6 +161,7 @@ public abstract class AbstractCanalAdapterWorker {
} }
logger.info("topic all connectors destroyed!"); logger.info("topic all connectors destroyed!");
} }
public abstract void start(); public abstract void start();
public abstract void stop(); public abstract void stop();

4
client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java
View File

@@ -30,7 +30,7 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
private boolean flatMessage; private boolean flatMessage;
public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId, public CanalAdapterKafkaWorker(String bootstrapServers, String topic, String groupId,
List<List<CanalOuterAdapter>> canalOuterAdapters, boolean flatMessage){ List<List<CanalOuterAdapter>> canalOuterAdapters, boolean flatMessage){
this.canalOuterAdapters = canalOuterAdapters; this.canalOuterAdapters = canalOuterAdapters;
this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
this.topic = topic; this.topic = topic;
@@ -171,4 +171,4 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector.disconnect(); connector.disconnect();
logger.info("=============> Disconnect topic: {} <=============", this.topic); logger.info("=============> Disconnect topic: {} <=============", this.topic);
} }
} }

44
client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java
View File

@@ -1,9 +1,5 @@
package com.alibaba.otter.canal.client.adapter.loader; package com.alibaba.otter.canal.client.adapter.loader;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader;
import java.net.InetSocketAddress; import java.net.InetSocketAddress;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.util.ArrayList; import java.util.ArrayList;
@@ -12,10 +8,16 @@ import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader;
/** /**
* MQ外部适配器的加载器 * MQ外部适配器的加载器
* *
@@ -23,17 +25,17 @@ import org.slf4j.LoggerFactory;
*/ */
public class CanalAdapterLoader { public class CanalAdapterLoader {
private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class); private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class);
private CanalClientConfig canalClientConfig; private CanalClientConfig canalClientConfig;
private Map<String, CanalAdapterWorker> canalWorkers = new HashMap<>(); private Map<String, CanalAdapterWorker> canalWorkers = new HashMap<>();
private Map<String, AbstractCanalAdapterWorker> canalMQWorker = new HashMap<>(); private Map<String, AbstractCanalAdapterWorker> canalMQWorker = new HashMap<>();
private ExtensionLoader<CanalOuterAdapter> loader; private ExtensionLoader<CanalOuterAdapter> loader;
public CanalAdapterLoader(CanalClientConfig canalClientConfig) { public CanalAdapterLoader(CanalClientConfig canalClientConfig){
this.canalClientConfig = canalClientConfig; this.canalClientConfig = canalClientConfig;
} }
@@ -46,8 +48,14 @@ public class CanalAdapterLoader {
throw new RuntimeException("Blank config property: canalInstances or canalMQTopics"); throw new RuntimeException("Blank config property: canalInstances or canalMQTopics");
} }
loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, "" /*
"" /* TODO canalClientConfig.getClassloaderPolicy() */); * TODO
* canalClientConfig
* .
* getClassloaderPolicy
* (
* )
*/);
String canalServerHost = this.canalClientConfig.getCanalServerHost(); String canalServerHost = this.canalClientConfig.getCanalServerHost();
SocketAddress sa = null; SocketAddress sa = null;
@@ -94,24 +102,23 @@ public class CanalAdapterLoader {
} }
canalOuterAdapterGroups.add(canalOuterAdapters); canalOuterAdapterGroups.add(canalOuterAdapters);
if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) { if (StringUtils.isBlank(topic.getMqMode()) || "rocketmq".equalsIgnoreCase(topic.getMqMode())) {
CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker( CanalAdapterRocketMQWorker rocketMQWorker = new CanalAdapterRocketMQWorker(canalClientConfig.getBootstrapServers(),
canalClientConfig.getBootstrapServers(),
topic.getTopic(), topic.getTopic(),
group.getGroupId(), group.getGroupId(),
canalOuterAdapterGroups); canalOuterAdapterGroups);
canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker); canalMQWorker.put(topic.getTopic() + "-rocketmq-" + group.getGroupId(), rocketMQWorker);
rocketMQWorker.start(); rocketMQWorker.start();
} else if ("kafka".equalsIgnoreCase(topic.getMqMode())) { } else if ("kafka".equalsIgnoreCase(topic.getMqMode())) {
CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker( CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(canalClientConfig.getBootstrapServers(),
canalClientConfig.getBootstrapServers(),
topic.getTopic(), topic.getTopic(),
group.getGroupId(), group.getGroupId(),
canalOuterAdapterGroups,canalClientConfig.getFlatMessage()); canalOuterAdapterGroups,
canalClientConfig.getFlatMessage());
canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker); canalMQWorker.put(topic.getTopic() + "-kafka-" + group.getGroupId(), canalKafkaWorker);
canalKafkaWorker.start(); canalKafkaWorker.start();
} }
logger.info("Start adapter for canal-client rocketmq topic: {} succeed", logger.info("Start adapter for canal-client rocketmq topic: {} succeed", topic.getTopic() + "-"
topic.getTopic() + "-" + group.getGroupId()); + group.getGroupId());
} }
} }
@@ -156,6 +163,7 @@ public class CanalAdapterLoader {
for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) { for (AbstractCanalAdapterWorker tmp : canalMQWorker.values()) {
final AbstractCanalAdapterWorker worker = tmp; final AbstractCanalAdapterWorker worker = tmp;
stopMQWokerService.submit(new Runnable() { stopMQWokerService.submit(new Runnable() {
@Override @Override
public void run() { public void run() {
worker.stop(); worker.stop();

19
client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterRocketMQWorker.java
View File

@@ -1,16 +1,17 @@
package com.alibaba.otter.canal.client.adapter.loader; package com.alibaba.otter.canal.client.adapter.loader;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.common.errors.WakeupException;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter; import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.common.errors.WakeupException;
/** /**
* kafka对应的client适配器工作线程 * kafka对应的client适配器工作线程
@@ -22,10 +23,10 @@ public class CanalAdapterRocketMQWorker extends AbstractCanalAdapterWorker {
private RocketMQCanalConnector connector; private RocketMQCanalConnector connector;
private String topic; private String topic;
public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId, public CanalAdapterRocketMQWorker(String nameServers, String topic, String groupId,
List<List<CanalOuterAdapter>> canalOuterAdapters) { List<List<CanalOuterAdapter>> canalOuterAdapters){
logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId); logger.info("RocketMQ consumer config topic:{}, nameServer:{}, groupId:{}", topic, nameServers, groupId);
this.canalOuterAdapters = canalOuterAdapters; this.canalOuterAdapters = canalOuterAdapters;
this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size()); this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());

37
client/src/main/java/com/alibaba/otter/canal/client/CanalMessageDeserializer.java
View File

@@ -2,16 +2,19 @@ package com.alibaba.otter.canal.client;
import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket; import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.CanalPacket.Ack;
import com.alibaba.otter.canal.protocol.CanalPacket.Compression;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException; import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.google.protobuf.ByteString; import com.google.protobuf.ByteString;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class CanalMessageDeserializer { public class CanalMessageDeserializer {
private static Logger logger = LoggerFactory.getLogger(CanalMessageDeserializer.class);
public static Message deserializer(byte[] data) { public static Message deserializer(byte[] data) {
return deserializer(data, false);
}
public static Message deserializer(byte[] data, boolean lazyParseEntry) {
try { try {
if (data == null) { if (data == null) {
return null; return null;
@@ -19,24 +22,34 @@ public class CanalMessageDeserializer {
CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data); CanalPacket.Packet p = CanalPacket.Packet.parseFrom(data);
switch (p.getType()) { switch (p.getType()) {
case MESSAGES: { case MESSAGES: {
// if (!p.getCompression().equals(CanalPacket.Compression.NONE)) { if (!p.getCompression().equals(Compression.NONE)
// throw new CanalClientException("compression is not supported in this connector"); && !p.getCompression().equals(Compression.COMPRESSIONCOMPATIBLEPROTO2)) {
// } throw new CanalClientException("compression is not supported in this connector");
}
CanalPacket.Messages messages = CanalPacket.Messages.parseFrom(p.getBody()); CanalPacket.Messages messages = CanalPacket.Messages.parseFrom(p.getBody());
Message result = new Message(messages.getBatchId()); Message result = new Message(messages.getBatchId());
for (ByteString byteString : messages.getMessagesList()) { if (lazyParseEntry) {
result.addEntry(CanalEntry.Entry.parseFrom(byteString)); // byteString
result.setRawEntries(messages.getMessagesList());
} else {
for (ByteString byteString : messages.getMessagesList()) {
result.addEntry(CanalEntry.Entry.parseFrom(byteString));
}
} }
return result; return result;
} }
default: case ACK: {
break; Ack ack = Ack.parseFrom(p.getBody());
throw new CanalClientException("something goes wrong with reason: " + ack.getErrorMessage());
}
default: {
throw new CanalClientException("unexpected packet type: " + p.getType());
}
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error("Error when deserializing byte[] to message ", e); throw new CanalClientException("deserializer failed", e);
} }
return null;
} }
} }

9
client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java
View File

@@ -218,8 +218,7 @@ public class ClusterCanalConnector implements CanalConnector {
return; return;
} catch (Throwable t) { } catch (Throwable t) {
logger.warn(String.format("something goes wrong when rollbacking data from server:%s", logger.warn(String.format("something goes wrong when rollbacking data from server:%s",
currentConnector != null ? currentConnector.getAddress() : "null"), currentConnector != null ? currentConnector.getAddress() : "null"), t);
t);
times++; times++;
restart(); restart();
logger.info("restart the connector for next round retry."); logger.info("restart the connector for next round retry.");
@@ -236,8 +235,7 @@ public class ClusterCanalConnector implements CanalConnector {
return; return;
} catch (Throwable t) { } catch (Throwable t) {
logger.warn(String.format("something goes wrong when rollbacking data from server:%s", logger.warn(String.format("something goes wrong when rollbacking data from server:%s",
currentConnector != null ? currentConnector.getAddress() : "null"), currentConnector != null ? currentConnector.getAddress() : "null"), t);
t);
times++; times++;
restart(); restart();
logger.info("restart the connector for next round retry."); logger.info("restart the connector for next round retry.");
@@ -255,8 +253,7 @@ public class ClusterCanalConnector implements CanalConnector {
return; return;
} catch (Throwable t) { } catch (Throwable t) {
logger.warn(String.format("something goes wrong when acking data from server:%s", logger.warn(String.format("something goes wrong when acking data from server:%s",
currentConnector != null ? currentConnector.getAddress() : "null"), currentConnector != null ? currentConnector.getAddress() : "null"), t);
t);
times++; times++;
restart(); restart();
logger.info("restart the connector for next round retry."); logger.info("restart the connector for next round retry.");

43
client/src/main/java/com/alibaba/otter/canal/client/impl/SimpleCanalConnector.java
View File

@@ -18,13 +18,13 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.client.impl.running.ClientRunningData; import com.alibaba.otter.canal.client.impl.running.ClientRunningData;
import com.alibaba.otter.canal.client.impl.running.ClientRunningListener; import com.alibaba.otter.canal.client.impl.running.ClientRunningListener;
import com.alibaba.otter.canal.client.impl.running.ClientRunningMonitor; import com.alibaba.otter.canal.client.impl.running.ClientRunningMonitor;
import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.common.utils.BooleanMutex; import com.alibaba.otter.canal.common.utils.BooleanMutex;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx; import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalPacket.Ack; import com.alibaba.otter.canal.protocol.CanalPacket.Ack;
import com.alibaba.otter.canal.protocol.CanalPacket.ClientAck; import com.alibaba.otter.canal.protocol.CanalPacket.ClientAck;
import com.alibaba.otter.canal.protocol.CanalPacket.ClientAuth; import com.alibaba.otter.canal.protocol.CanalPacket.ClientAuth;
@@ -32,7 +32,6 @@ import com.alibaba.otter.canal.protocol.CanalPacket.ClientRollback;
import com.alibaba.otter.canal.protocol.CanalPacket.Compression; import com.alibaba.otter.canal.protocol.CanalPacket.Compression;
import com.alibaba.otter.canal.protocol.CanalPacket.Get; import com.alibaba.otter.canal.protocol.CanalPacket.Get;
import com.alibaba.otter.canal.protocol.CanalPacket.Handshake; import com.alibaba.otter.canal.protocol.CanalPacket.Handshake;
import com.alibaba.otter.canal.protocol.CanalPacket.Messages;
import com.alibaba.otter.canal.protocol.CanalPacket.Packet; import com.alibaba.otter.canal.protocol.CanalPacket.Packet;
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType; import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
import com.alibaba.otter.canal.protocol.CanalPacket.Sub; import com.alibaba.otter.canal.protocol.CanalPacket.Sub;
@@ -180,8 +179,8 @@ public class SimpleCanalConnector implements CanalConnector {
Ack ackBody = Ack.parseFrom(ack.getBody()); Ack ackBody = Ack.parseFrom(ack.getBody());
if (ackBody.getErrorCode() > 0) { if (ackBody.getErrorCode() > 0) {
throw new CanalClientException( throw new CanalClientException("something goes wrong when doing authentication: "
"something goes wrong when doing authentication: " + ackBody.getErrorMessage()); + ackBody.getErrorMessage());
} }
connected = true; connected = true;
@@ -320,34 +319,7 @@ public class SimpleCanalConnector implements CanalConnector {
private Message receiveMessages() throws IOException { private Message receiveMessages() throws IOException {
byte[] data = readNextPacket(); byte[] data = readNextPacket();
Packet p = Packet.parseFrom(data); return CanalMessageDeserializer.deserializer(data, lazyParseEntry);
switch (p.getType()) {
case MESSAGES: {
if (!p.getCompression().equals(Compression.NONE)
&& !p.getCompression().equals(Compression.COMPRESSIONCOMPATIBLEPROTO2)) {
throw new CanalClientException("compression is not supported in this connector");
}
Messages messages = Messages.parseFrom(p.getBody());
Message result = new Message(messages.getBatchId());
if (lazyParseEntry) {
// byteString
result.setRawEntries(messages.getMessagesList());
} else {
for (ByteString byteString : messages.getMessagesList()) {
result.addEntry(Entry.parseFrom(byteString));
}
}
return result;
}
case ACK: {
Ack ack = Ack.parseFrom(p.getBody());
throw new CanalClientException("something goes wrong with reason: " + ack.getErrorMessage());
}
default: {
throw new CanalClientException("unexpected packet type: " + p.getType());
}
}
} }
public void ack(long batchId) throws CanalClientException { public void ack(long batchId) throws CanalClientException {
@@ -361,8 +333,11 @@ public class SimpleCanalConnector implements CanalConnector {
.setBatchId(batchId) .setBatchId(batchId)
.build(); .build();
try { try {
writeWithHeader( writeWithHeader(Packet.newBuilder()
Packet.newBuilder().setType(PacketType.CLIENTACK).setBody(ca.toByteString()).build().toByteArray()); .setType(PacketType.CLIENTACK)
.setBody(ca.toByteString())
.build()
.toByteArray());
} catch (IOException e) { } catch (IOException e) {
throw new CanalClientException(e); throw new CanalClientException(e);
} }

2
client/src/main/java/com/alibaba/otter/canal/client/impl/running/ClientRunningMonitor.java
View File

@@ -98,7 +98,7 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
String path = ZookeeperPathUtils.getDestinationClientRunning(this.destination, clientData.getClientId()); String path = ZookeeperPathUtils.getDestinationClientRunning(this.destination, clientData.getClientId());
zkClient.unsubscribeDataChanges(path, dataListener); zkClient.unsubscribeDataChanges(path, dataListener);
releaseRunning(); // 尝试一下release releaseRunning(); // 尝试一下release
//Fix issue #697 // Fix issue #697
if (delayExector != null) { if (delayExector != null) {
delayExector.shutdown(); delayExector.shutdown();
} }

6
client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
View File

@@ -13,6 +13,7 @@ import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringDeserializer;
import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
@@ -33,8 +34,7 @@ public class KafkaCanalConnector {
private volatile boolean running = false; private volatile boolean running = false;
private boolean flatMessage; private boolean flatMessage;
public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, boolean flatMessage){
boolean flatMessage){
this.topic = topic; this.topic = topic;
this.partition = partition; this.partition = partition;
this.flatMessage = flatMessage; this.flatMessage = flatMessage;
@@ -50,7 +50,7 @@ public class KafkaCanalConnector {
properties.put("max.poll.records", "100"); properties.put("max.poll.records", "100");
properties.put("key.deserializer", StringDeserializer.class.getName()); properties.put("key.deserializer", StringDeserializer.class.getName());
if (!flatMessage) { if (!flatMessage) {
properties.put("value.deserializer", MessageDeserializer.class.getName()); properties.put("value.deserializer", CanalMessageDeserializer.class.getName());
} else { } else {
properties.put("value.deserializer", StringDeserializer.class.getName()); properties.put("value.deserializer", StringDeserializer.class.getName());
} }

7
client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnectors.java
View File

@@ -17,8 +17,7 @@ public class KafkaCanalConnectors {
* @param groupId * @param groupId
* @return * @return
*/ */
public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition, String groupId) {
String groupId) {
return new KafkaCanalConnector(servers, topic, partition, groupId, false); return new KafkaCanalConnector(servers, topic, partition, groupId, false);
} }
@@ -44,8 +43,8 @@ public class KafkaCanalConnectors {
* @param flatMessage * @param flatMessage
* @return * @return
*/ */
public static KafkaCanalConnector newKafkaConnector( String servers, String topic, public static KafkaCanalConnector newKafkaConnector(String servers, String topic, Integer partition,
Integer partition, String groupId,boolean flatMessage) { String groupId, boolean flatMessage) {
return new KafkaCanalConnector(servers, topic, partition, groupId, flatMessage); return new KafkaCanalConnector(servers, topic, partition, groupId, flatMessage);
} }
} }

33
client/src/main/java/com/alibaba/otter/canal/client/kafka/MessageDeserializer.java
View File

@@ -1,33 +0,0 @@
package com.alibaba.otter.canal.client.kafka;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.protocol.Message;
import java.util.Map;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Kafka Message类的反序列化
*
* @author machengyuan @ 2018-6-12
* @version 1.0.0
*/
public class MessageDeserializer implements Deserializer<Message> {
private static Logger logger = LoggerFactory.getLogger(MessageDeserializer.class);
@Override
public void configure(Map<String, ?> configs, boolean isKey) {
}
@Override
public Message deserialize(String topic1, byte[] data) {
return CanalMessageDeserializer.deserializer(data);
}
@Override
public void close() {
// nothing to do
}
}

9
client/src/main/java/com/alibaba/otter/canal/client/kafka/running/ClientRunningMonitor.java
View File

@@ -24,7 +24,6 @@ import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils; import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
import com.alibaba.otter.canal.protocol.exception.CanalClientException; import com.alibaba.otter.canal.protocol.exception.CanalClientException;
/** /**
* kafka client running状态信息 * kafka client running状态信息
* *
@@ -55,15 +54,15 @@ public class ClientRunningMonitor extends AbstractCanalLifeCycle {
} }
private static final Logger logger = LoggerFactory.getLogger(ClientRunningMonitor.class); private static final Logger logger = LoggerFactory.getLogger(ClientRunningMonitor.class);
private ZkClientx zkClient; private ZkClientx zkClient;
private String topic; private String topic;
private ClientRunningData clientData; private ClientRunningData clientData;
private IZkDataListener dataListener; private IZkDataListener dataListener;
private BooleanMutex mutex = new BooleanMutex(false); private BooleanMutex mutex = new BooleanMutex(false);
private volatile boolean release = false; private volatile boolean release = false;
private volatile ClientRunningData activeData; private volatile ClientRunningData activeData;
private ScheduledExecutorService delayExector = Executors.newScheduledThreadPool(1); private ScheduledExecutorService delayExector = Executors.newScheduledThreadPool(1);
private ClientRunningListener listener; private ClientRunningListener listener;
private int delayTime = 5; private int delayTime = 5;
private static Integer virtualPort; private static Integer virtualPort;

9
client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java
View File

@@ -5,11 +5,12 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
public class ConsumerBatchMessage<T> { public class ConsumerBatchMessage<T> {
private final BlockingQueue<T> data;
private CountDownLatch latch;
private boolean hasFailure = false;
public ConsumerBatchMessage(BlockingQueue<T> data) { private final BlockingQueue<T> data;
private CountDownLatch latch;
private boolean hasFailure = false;
public ConsumerBatchMessage(BlockingQueue<T> data){
this.data = data; this.data = data;
latch = new CountDownLatch(data.size()); latch = new CountDownLatch(data.size());
} }

33
client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java
View File

@@ -1,15 +1,12 @@
package com.alibaba.otter.canal.client.rocketmq; package com.alibaba.otter.canal.client.rocketmq;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils; import org.apache.commons.lang.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext; import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
@@ -20,19 +17,25 @@ import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalMessageDeserializer;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
public class RocketMQCanalConnector implements CanalConnector { public class RocketMQCanalConnector implements CanalConnector {
private static final Logger logger = LoggerFactory.getLogger(RocketMQCanalConnector.class);
private String nameServer; private static final Logger logger = LoggerFactory.getLogger(RocketMQCanalConnector.class);
private String topic;
private String groupName; private String nameServer;
private volatile boolean connected = false; private String topic;
private DefaultMQPushConsumer rocketMQConsumer; private String groupName;
private volatile boolean connected = false;
private DefaultMQPushConsumer rocketMQConsumer;
private BlockingQueue<ConsumerBatchMessage<Message>> messageBlockingQueue; private BlockingQueue<ConsumerBatchMessage<Message>> messageBlockingQueue;
Map<Long, ConsumerBatchMessage<Message>> messageCache; Map<Long, ConsumerBatchMessage<Message>> messageCache;
private long batchProcessTimeout = 10000; private long batchProcessTimeout = 10000;
public RocketMQCanalConnector(String nameServer, String topic, String groupName) { public RocketMQCanalConnector(String nameServer, String topic, String groupName){
this.nameServer = nameServer; this.nameServer = nameServer;
this.topic = topic; this.topic = topic;
this.groupName = groupName; this.groupName = groupName;
@@ -69,9 +72,9 @@ public class RocketMQCanalConnector implements CanalConnector {
} }
rocketMQConsumer.subscribe(topic, "*"); rocketMQConsumer.subscribe(topic, "*");
rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() { rocketMQConsumer.registerMessageListener(new MessageListenerOrderly() {
@Override @Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts, public ConsumeOrderlyStatus consumeMessage(List<MessageExt> messageExts, ConsumeOrderlyContext context) {
ConsumeOrderlyContext context) {
context.setAutoCommit(true); context.setAutoCommit(true);
boolean isSuccess = process(messageExts); boolean isSuccess = process(messageExts);
if (isSuccess) { if (isSuccess) {

4
client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnectorProvider.java
View File

@@ -4,6 +4,7 @@ package com.alibaba.otter.canal.client.rocketmq;
* RocketMQ connector provider. * RocketMQ connector provider.
*/ */
public class RocketMQCanalConnectorProvider { public class RocketMQCanalConnectorProvider {
/** /**
* Create RocketMQ connector * Create RocketMQ connector
* *
@@ -12,8 +13,7 @@ public class RocketMQCanalConnectorProvider {
* @param groupId * @param groupId
* @return {@link RocketMQCanalConnector} * @return {@link RocketMQCanalConnector}
*/ */
public static RocketMQCanalConnector newRocketMQConnector(String nameServers, String topic, public static RocketMQCanalConnector newRocketMQConnector(String nameServers, String topic, String groupId) {
String groupId) {
return new RocketMQCanalConnector(nameServers, topic, groupId); return new RocketMQCanalConnector(nameServers, topic, groupId);
} }

2
client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalListener.java
View File

@@ -1,11 +1,13 @@
package com.alibaba.otter.canal.client.rocketmq; package com.alibaba.otter.canal.client.rocketmq;
import java.util.List; import java.util.List;
import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageExt;
/** /**
* RocketMQ message listener * RocketMQ message listener
*/ */
public interface RocketMQCanalListener { public interface RocketMQCanalListener {
boolean onReceive(List<MessageExt> messageExts); boolean onReceive(List<MessageExt> messageExts);
} }

188
client/src/test/java/com/alibaba/otter/canal/client/running/ClientRunningTest.java
View File

@@ -20,117 +20,117 @@ import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
public class ClientRunningTest extends AbstractZkTest { public class ClientRunningTest extends AbstractZkTest {
private ZkClientx zkclientx = new ZkClientx(cluster1 + ";" + cluster2); private ZkClientx zkclientx = new ZkClientx(cluster1 + ";" + cluster2);
private short clientId = 1001; private short clientId = 1001;
@Before @Before
public void setUp() { public void setUp() {
String path = ZookeeperPathUtils.getDestinationPath(destination); String path = ZookeeperPathUtils.getDestinationPath(destination);
zkclientx.deleteRecursive(path); zkclientx.deleteRecursive(path);
zkclientx.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientId), true); zkclientx.createPersistent(ZookeeperPathUtils.getClientIdNodePath(this.destination, clientId), true);
} }
@After @After
public void tearDown() { public void tearDown() {
String path = ZookeeperPathUtils.getDestinationPath(destination); String path = ZookeeperPathUtils.getDestinationPath(destination);
zkclientx.deleteRecursive(path); zkclientx.deleteRecursive(path);
} }
@Test @Test
public void testOneServer() { public void testOneServer() {
final CountDownLatch countLatch = new CountDownLatch(2); final CountDownLatch countLatch = new CountDownLatch(2);
ClientRunningMonitor runningMonitor = buildClientRunning(countLatch, clientId, 2088); ClientRunningMonitor runningMonitor = buildClientRunning(countLatch, clientId, 2088);
runningMonitor.start(); runningMonitor.start();
sleep(2000L); sleep(2000L);
runningMonitor.stop(); runningMonitor.stop();
sleep(2000L); sleep(2000L);
if (countLatch.getCount() != 0) { if (countLatch.getCount() != 0) {
Assert.fail(); Assert.fail();
} }
} }
@Test @Test
public void testMultiServer() { public void testMultiServer() {
final CountDownLatch countLatch = new CountDownLatch(30); final CountDownLatch countLatch = new CountDownLatch(30);
final ClientRunningMonitor runningMonitor1 = buildClientRunning(countLatch, clientId, 2088); final ClientRunningMonitor runningMonitor1 = buildClientRunning(countLatch, clientId, 2088);
final ClientRunningMonitor runningMonitor2 = buildClientRunning(countLatch, clientId, 2089); final ClientRunningMonitor runningMonitor2 = buildClientRunning(countLatch, clientId, 2089);
final ClientRunningMonitor runningMonitor3 = buildClientRunning(countLatch, clientId, 2090); final ClientRunningMonitor runningMonitor3 = buildClientRunning(countLatch, clientId, 2090);
final ExecutorService executor = Executors.newFixedThreadPool(3); final ExecutorService executor = Executors.newFixedThreadPool(3);
executor.submit(new Runnable() { executor.submit(new Runnable() {
public void run() { public void run() {
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
if (!runningMonitor1.isStart()) { if (!runningMonitor1.isStart()) {
runningMonitor1.start(); runningMonitor1.start();
} }
sleep(2000L + RandomUtils.nextInt(500)); sleep(2000L + RandomUtils.nextInt(500));
runningMonitor1.stop(); runningMonitor1.stop();
sleep(2000L + RandomUtils.nextInt(500)); sleep(2000L + RandomUtils.nextInt(500));
} }
} }
}); });
executor.submit(new Runnable() { executor.submit(new Runnable() {
public void run() { public void run() {
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
if (!runningMonitor2.isStart()) { if (!runningMonitor2.isStart()) {
runningMonitor2.start(); runningMonitor2.start();
} }
sleep(2000L + RandomUtils.nextInt(500)); sleep(2000L + RandomUtils.nextInt(500));
runningMonitor2.stop(); runningMonitor2.stop();
sleep(2000L + RandomUtils.nextInt(500)); sleep(2000L + RandomUtils.nextInt(500));
} }
} }
}); });
executor.submit(new Runnable() { executor.submit(new Runnable() {
public void run() { public void run() {
for (int i = 0; i < 10; i++) { for (int i = 0; i < 10; i++) {
if (!runningMonitor3.isStart()) { if (!runningMonitor3.isStart()) {
runningMonitor3.start(); runningMonitor3.start();
} }
sleep(2000L + RandomUtils.nextInt(500)); sleep(2000L + RandomUtils.nextInt(500));
runningMonitor3.stop(); runningMonitor3.stop();
sleep(2000L + RandomUtils.nextInt(500)); sleep(2000L + RandomUtils.nextInt(500));
} }
} }
}); });
sleep(30000L); sleep(30000L);
} }
private ClientRunningMonitor buildClientRunning(final CountDownLatch countLatch, final short clientId, private ClientRunningMonitor buildClientRunning(final CountDownLatch countLatch, final short clientId,
final int port) { final int port) {
ClientRunningData clientData = new ClientRunningData(); ClientRunningData clientData = new ClientRunningData();
clientData.setClientId(clientId); clientData.setClientId(clientId);
clientData.setAddress(AddressUtils.getHostIp()); clientData.setAddress(AddressUtils.getHostIp());
ClientRunningMonitor runningMonitor = new ClientRunningMonitor(); ClientRunningMonitor runningMonitor = new ClientRunningMonitor();
runningMonitor.setDestination(destination); runningMonitor.setDestination(destination);
runningMonitor.setZkClient(zkclientx); runningMonitor.setZkClient(zkclientx);
runningMonitor.setClientData(clientData); runningMonitor.setClientData(clientData);
runningMonitor.setListener(new ClientRunningListener() { runningMonitor.setListener(new ClientRunningListener() {
public InetSocketAddress processActiveEnter() { public InetSocketAddress processActiveEnter() {
System.out.println(String.format("clientId:%s port:%s has start", clientId, port)); System.out.println(String.format("clientId:%s port:%s has start", clientId, port));
countLatch.countDown(); countLatch.countDown();
return new InetSocketAddress(AddressUtils.getHostIp(), port); return new InetSocketAddress(AddressUtils.getHostIp(), port);
} }
public void processActiveExit() { public void processActiveExit() {
countLatch.countDown(); countLatch.countDown();
System.out.println(String.format("clientId:%s port:%s has stop", clientId, port)); System.out.println(String.format("clientId:%s port:%s has stop", clientId, port));
} }
}); });
runningMonitor.setDelayTime(1); runningMonitor.setDelayTime(1);
return runningMonitor; return runningMonitor;
} }
} }

3
client/src/test/java/com/alibaba/otter/canal/client/running/kafka/CanalKafkaClientExample.java
View File

@@ -41,8 +41,7 @@ public class CanalKafkaClientExample {
public static void main(String[] args) { public static void main(String[] args) {
try { try {
final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample( final CanalKafkaClientExample kafkaCanalClientExample = new CanalKafkaClientExample(AbstractKafkaTest.zkServers,
AbstractKafkaTest.zkServers,
AbstractKafkaTest.servers, AbstractKafkaTest.servers,
AbstractKafkaTest.topic, AbstractKafkaTest.topic,
AbstractKafkaTest.partition, AbstractKafkaTest.partition,

5
client/src/test/java/com/alibaba/otter/canal/client/running/rocketmq/AbstractRocektMQTest.java
View File

@@ -1,8 +1,9 @@
package com.alibaba.otter.canal.client.running.rocketmq; package com.alibaba.otter.canal.client.running.rocketmq;
public class AbstractRocektMQTest { public class AbstractRocektMQTest {
public static String topic = "example";
public static String groupId = "group"; public static String topic = "example";
public static String groupId = "group";
public static String nameServers = "localhost:9876"; public static String nameServers = "localhost:9876";
} }

35
client/src/test/java/com/alibaba/otter/canal/client/running/rocketmq/CanalRocketMQClientExample.java
View File

@@ -1,13 +1,14 @@
package com.alibaba.otter.canal.client.running.rocketmq; package com.alibaba.otter.canal.client.running.rocketmq;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnector;
import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider; import com.alibaba.otter.canal.client.rocketmq.RocketMQCanalConnectorProvider;
import com.alibaba.otter.canal.client.running.kafka.AbstractKafkaTest; import com.alibaba.otter.canal.client.running.kafka.AbstractKafkaTest;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import org.apache.kafka.common.errors.WakeupException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.Assert;
/** /**
* Kafka client example * Kafka client example
@@ -17,32 +18,30 @@ import org.springframework.util.Assert;
*/ */
public class CanalRocketMQClientExample extends AbstractRocektMQTest { public class CanalRocketMQClientExample extends AbstractRocektMQTest {
protected final static Logger logger = LoggerFactory.getLogger(CanalRocketMQClientExample.class); protected final static Logger logger = LoggerFactory.getLogger(CanalRocketMQClientExample.class);
private RocketMQCanalConnector connector; private RocketMQCanalConnector connector;
private static volatile boolean running = false; private static volatile boolean running = false;
private Thread thread = null; private Thread thread = null;
private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() { private Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
public void uncaughtException(Thread t, Throwable e) { public void uncaughtException(Thread t, Throwable e) {
logger.error("parse events has an error", e); logger.error("parse events has an error", e);
} }
}; };
public CanalRocketMQClientExample(String nameServers, String topic, public CanalRocketMQClientExample(String nameServers, String topic, String groupId){
String groupId) { connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers, topic, groupId);
connector = RocketMQCanalConnectorProvider.newRocketMQConnector(nameServers,
topic,
groupId);
} }
public static void main(String[] args) { public static void main(String[] args) {
try { try {
final CanalRocketMQClientExample rocketMQClientExample = new CanalRocketMQClientExample(nameServers, final CanalRocketMQClientExample rocketMQClientExample = new CanalRocketMQClientExample(nameServers,
topic, groupId); topic,
groupId);
logger.info("## Start the rocketmq consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId); logger.info("## Start the rocketmq consumer: {}-{}", AbstractKafkaTest.topic, AbstractKafkaTest.groupId);
rocketMQClientExample.start(); rocketMQClientExample.start();
logger.info("## The canal rocketmq consumer is running now ......"); logger.info("## The canal rocketmq consumer is running now ......");

2
deployer/src/main/resources/canal.properties
View File

@@ -10,7 +10,7 @@ canal.zkServers=
canal.zookeeper.flush.period = 1000 canal.zookeeper.flush.period = 1000
canal.withoutNetty = false canal.withoutNetty = false
# tcp, kafka, RocketMQ # tcp, kafka, RocketMQ
canal.serverMode = rocketmq canal.serverMode = tcp
# flush meta cursor/parse position to file # flush meta cursor/parse position to file
canal.file.data.dir = ${canal.conf.dir} canal.file.data.dir = ${canal.conf.dir}
canal.file.flush.period = 1000 canal.file.flush.period = 1000

7
protocol/src/main/java/com/alibaba/otter/canal/protocol/ClientIdentity.java
View File

@@ -13,9 +13,10 @@ import com.alibaba.otter.canal.common.utils.CanalToStringStyle;
*/ */
public class ClientIdentity implements Serializable { public class ClientIdentity implements Serializable {
private String destination; private static final long serialVersionUID = -8262100681930834834L;
private short clientId; private String destination;
private String filter; private short clientId;
private String filter;
public ClientIdentity(){ public ClientIdentity(){

18
server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java
View File

@@ -1,5 +1,10 @@
package com.alibaba.otter.canal.common; package com.alibaba.otter.canal.common;
import java.util.List;
import org.apache.kafka.common.errors.SerializationException;
import org.springframework.util.CollectionUtils;
import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.CanalPacket; import com.alibaba.otter.canal.protocol.CanalPacket;
import com.alibaba.otter.canal.protocol.CanalPacket.PacketType; import com.alibaba.otter.canal.protocol.CanalPacket.PacketType;
@@ -7,12 +12,11 @@ import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.ByteString; import com.google.protobuf.ByteString;
import com.google.protobuf.CodedOutputStream; import com.google.protobuf.CodedOutputStream;
import com.google.protobuf.WireFormat; import com.google.protobuf.WireFormat;
import java.util.List;
import org.apache.kafka.common.errors.SerializationException;
import org.springframework.util.CollectionUtils;
public class CanalMessageSerializer { public class CanalMessageSerializer {
public static byte[] serializer(Message data){
@SuppressWarnings("deprecation")
public static byte[] serializer(Message data) {
try { try {
if (data != null) { if (data != null) {
if (data.getId() != -1) { if (data.getId() != -1) {
@@ -31,11 +35,9 @@ public class CanalMessageSerializer {
messageSize += 1 * rowEntries.size(); messageSize += 1 * rowEntries.size();
// packet size // packet size
int size = 0; int size = 0;
size += CodedOutputStream.computeEnumSize(3, size += CodedOutputStream.computeEnumSize(3, PacketType.MESSAGES.getNumber());
PacketType.MESSAGES.getNumber());
size += CodedOutputStream.computeTagSize(5) size += CodedOutputStream.computeTagSize(5)
+ CodedOutputStream.computeRawVarint32Size(messageSize) + CodedOutputStream.computeRawVarint32Size(messageSize) + messageSize;
+ messageSize;
// build data // build data
byte[] body = new byte[size]; byte[] body = new byte[size];
CodedOutputStream output = CodedOutputStream.newInstance(body); CodedOutputStream output = CodedOutputStream.newInstance(body);

31
server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
View File

@@ -12,25 +12,25 @@ import java.util.Map;
*/ */
public class MQProperties { public class MQProperties {
private String servers = "localhost:6667"; private String servers = "localhost:6667";
private int retries = 0; private int retries = 0;
private int batchSize = 16384; private int batchSize = 16384;
private int lingerMs = 1; private int lingerMs = 1;
private long bufferMemory = 33554432L; private long bufferMemory = 33554432L;
private boolean filterTransactionEntry = true; private boolean filterTransactionEntry = true;
private String producerGroup = "Canal-Producer"; private String producerGroup = "Canal-Producer";
private int canalBatchSize = 50; private int canalBatchSize = 50;
private Long canalGetTimeout; private Long canalGetTimeout;
private boolean flatMessage = true; private boolean flatMessage = true;
private List<CanalDestination> canalDestinations = new ArrayList<CanalDestination>(); private List<CanalDestination> canalDestinations = new ArrayList<CanalDestination>();
public static class CanalDestination { public static class CanalDestination {
private String canalDestination; private String canalDestination;
private String topic; private String topic;
private Integer partition; private Integer partition;
private Integer partitionsNum; private Integer partitionsNum;
private Map<String, String> partitionHash; private Map<String, String> partitionHash;
public String getCanalDestination() { public String getCanalDestination() {
@@ -74,7 +74,6 @@ public class MQProperties {
} }
} }
public String getServers() { public String getServers() {
return servers; return servers;
} }

31
server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java
View File

@@ -1,12 +1,8 @@
package com.alibaba.otter.canal.kafka; package com.alibaba.otter.canal.kafka;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.spi.CanalMQProducer;
import java.util.List; import java.util.List;
import java.util.Properties; import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.ProducerRecord;
@@ -14,6 +10,13 @@ import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.spi.CanalMQProducer;
;
/** /**
* kafka producer 主操作类 * kafka producer 主操作类
* *
@@ -22,13 +25,13 @@ import org.slf4j.LoggerFactory;
*/ */
public class CanalKafkaProducer implements CanalMQProducer { public class CanalKafkaProducer implements CanalMQProducer {
private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class); private static final Logger logger = LoggerFactory.getLogger(CanalKafkaProducer.class);
private Producer<String, Message> producer; private Producer<String, Message> producer;
private Producer<String, String> producer2; // 用于扁平message的数据投递 private Producer<String, String> producer2; // 用于扁平message的数据投递
private MQProperties kafkaProperties; private MQProperties kafkaProperties;
@Override @Override
public void init(MQProperties kafkaProperties) { public void init(MQProperties kafkaProperties) {
@@ -91,8 +94,10 @@ public class CanalKafkaProducer implements CanalMQProducer {
if (flatMessages != null) { if (flatMessages != null) {
for (FlatMessage flatMessage : flatMessages) { for (FlatMessage flatMessage : flatMessages) {
if (canalDestination.getPartition() != null) { if (canalDestination.getPartition() != null) {
ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination.getTopic(),
.getTopic(), canalDestination.getPartition(), null, JSON.toJSONString(flatMessage)); canalDestination.getPartition(),
null,
JSON.toJSONString(flatMessage));
producer2.send(record); producer2.send(record);
} else { } else {
if (canalDestination.getPartitionHash() != null if (canalDestination.getPartitionHash() != null
@@ -104,8 +109,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
for (int i = 0; i < length; i++) { for (int i = 0; i < length; i++) {
FlatMessage flatMessagePart = partitionFlatMessage[i]; FlatMessage flatMessagePart = partitionFlatMessage[i];
if (flatMessagePart != null) { if (flatMessagePart != null) {
ProducerRecord<String, String> record = new ProducerRecord<String, String>( ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination.getTopic(),
canalDestination.getTopic(),
i, i,
null, null,
JSON.toJSONString(flatMessagePart)); JSON.toJSONString(flatMessagePart));
@@ -113,8 +117,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
} }
} }
} else { } else {
ProducerRecord<String, String> record = new ProducerRecord<String, String>( ProducerRecord<String, String> record = new ProducerRecord<String, String>(canalDestination.getTopic(),
canalDestination.getTopic(),
0, 0,
null, null,
JSON.toJSONString(flatMessage)); JSON.toJSONString(flatMessage));

6
server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java
View File

@@ -1,9 +1,11 @@
package com.alibaba.otter.canal.kafka; package com.alibaba.otter.canal.kafka;
import java.util.Map;
import org.apache.kafka.common.serialization.Serializer;
import com.alibaba.otter.canal.common.CanalMessageSerializer; import com.alibaba.otter.canal.common.CanalMessageSerializer;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import java.util.Map;
import org.apache.kafka.common.serialization.Serializer;
/** /**
* Kafka Message类的序列化 * Kafka Message类的序列化

6
server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
View File

@@ -16,9 +16,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
public class CanalRocketMQProducer implements CanalMQProducer { public class CanalRocketMQProducer implements CanalMQProducer {
private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class); private static final Logger logger = LoggerFactory.getLogger(CanalRocketMQProducer.class);
private DefaultMQProducer defaultMQProducer; private DefaultMQProducer defaultMQProducer;
@Override @Override
public void init(MQProperties rocketMQProperties) { public void init(MQProperties rocketMQProperties) {
@@ -36,10 +37,11 @@ public class CanalRocketMQProducer implements CanalMQProducer {
@Override @Override
public void send(final MQProperties.CanalDestination destination, com.alibaba.otter.canal.protocol.Message data, public void send(final MQProperties.CanalDestination destination, com.alibaba.otter.canal.protocol.Message data,
Callback callback) { Callback callback) {
try { try {
Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data)); Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data));
this.defaultMQProducer.send(message, new MessageQueueSelector() { this.defaultMQProducer.send(message, new MessageQueueSelector() {
@Override @Override
public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
int partition = 0; int partition = 0;

35
server/src/main/java/com/alibaba/otter/canal/server/CanalMQStarter.java
View File

@@ -1,35 +1,38 @@
package com.alibaba.otter.canal.server; package com.alibaba.otter.canal.server;
import java.io.FileInputStream;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.kafka.CanalKafkaProducer; import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
import com.alibaba.otter.canal.protocol.ClientIdentity; import com.alibaba.otter.canal.protocol.ClientIdentity;
import com.alibaba.otter.canal.protocol.Message; import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded; import com.alibaba.otter.canal.server.embedded.CanalServerWithEmbedded;
import com.alibaba.otter.canal.spi.CanalMQProducer; import com.alibaba.otter.canal.spi.CanalMQProducer;
import java.io.FileInputStream;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.yaml.snakeyaml.Yaml;
public class CanalMQStarter { public class CanalMQStarter {
private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class);
private static final Logger logger = LoggerFactory.getLogger(CanalMQStarter.class);
private static final String CLASSPATH_URL_PREFIX = "classpath:"; private static final String CLASSPATH_URL_PREFIX = "classpath:";
private volatile boolean running = false; private volatile boolean running = false;
private ExecutorService executorService; private ExecutorService executorService;
private CanalMQProducer canalMQProducer; private CanalMQProducer canalMQProducer;
private MQProperties properties; private MQProperties properties;
public CanalMQStarter(CanalMQProducer canalMQProducer) { public CanalMQStarter(CanalMQProducer canalMQProducer){
this.canalMQProducer = canalMQProducer; this.canalMQProducer = canalMQProducer;
} }
@@ -47,7 +50,7 @@ public class CanalMQStarter {
} }
// 初始化 kafka producer // 初始化 kafka producer
// canalMQProducer = new CanalKafkaProducer(); // canalMQProducer = new CanalKafkaProducer();
canalMQProducer.init(properties); canalMQProducer.init(properties);
// set filterTransactionEntry // set filterTransactionEntry
if (properties.isFilterTransactionEntry()) { if (properties.isFilterTransactionEntry()) {

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

@@ -50,7 +50,7 @@ public class SessionHandler extends SimpleChannelHandler {
this.embeddedServer = embeddedServer; this.embeddedServer = embeddedServer;
} }
@SuppressWarnings("deprecation") @SuppressWarnings({ "deprecation", "deprecation" })
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
logger.info("message receives in session handler..."); logger.info("message receives in session handler...");
long start = System.nanoTime(); long start = System.nanoTime();

9
server/src/main/java/com/alibaba/otter/canal/spi/CanalMQProducer.java
View File

@@ -1,11 +1,12 @@
package com.alibaba.otter.canal.spi; package com.alibaba.otter.canal.spi;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.kafka.CanalKafkaProducer;
import com.alibaba.otter.canal.protocol.Message;
import java.io.IOException; import java.io.IOException;
import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.Message;
public interface CanalMQProducer { public interface CanalMQProducer {
/** /**
* Init producer. * Init producer.
* *
@@ -33,4 +34,4 @@ public interface CanalMQProducer {
void rollback(); void rollback();
} }
} }