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

规范注释

Browse Source
This commit is contained in:
mcy
2018-10-25 21:21:06 +08:00
parent ddfc2464f3
commit d70dc8b88c
29 changed files with 230 additions and 132 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/OuterAdapter.java
View File

@@ -11,7 +11,7 @@ import com.alibaba.otter.canal.client.adapter.support.SPI;
/**
* 外部适配器接口
*
* @author machengyuan 2018-8-18 下午10:14:02
* @author reweerma 2018-8-18 下午10:14:02
* @version 1.0.0
*/
@SPI("logger")

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

@@ -1,13 +1,26 @@
package com.alibaba.otter.canal.client.adapter.support;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
/**
* 适配器配置集合, 用于配置加载, 线程不安全
*
* @author rewerma @ 2018-10-20
* @version 1.0.0
*/
public class AdapterConfigs {
private static Map<String, Set<String>> configs = new ConcurrentHashMap<>();
/**
* 类型下对应所有配置名, 如:
* hbase
* ┗━ mytest_person.yml
* ┗━ mytest_role.yml
* ┗━ mytest_department.yml
*/
private static Map<String, Set<String>> configs = new HashMap<>();
public static void put(String key, String value) {
Set<String> values = configs.get(key);

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

@@ -2,29 +2,26 @@ package com.alibaba.otter.canal.client.adapter.support;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
/**
* 配置信息类
*
* @author machengyuan 2018-8-18 下午10:40:12
* @author rewerma 2018-8-18 下午10:40:12
* @version 1.0.0
*/
public class CanalClientConfig {
private String canalServerHost;
private String canalServerHost; // 单机模式下canal server的 ip:port
private String zookeeperHosts;
private String zookeeperHosts; // 集群模式下的zk地址, 如果配置了单机地址则以单机为准!!
private Properties properties;
private String bootstrapServers; // kafka or rocket mq 地址
private String bootstrapServers;
private Boolean flatMessage = true; // 是否已flatMessage模式传输, 只适用于mq模式
private List<MQTopic> mqTopics;
private List<MQTopic> mqTopics; // mq topic 列表
private Boolean flatMessage = true;
private List<CanalInstance> canalInstances;
private List<CanalInstance> canalInstances; // tcp 模式下 canal 实例列表, 与mq模式不能共存!!
public String getCanalServerHost() {
return canalServerHost;
@@ -42,14 +39,6 @@ public class CanalClientConfig {
this.zookeeperHosts = zookeeperHosts;
}
public Properties getProperties() {
return properties;
}
public void setProperties(Properties properties) {
this.properties = properties;
}
public String getBootstrapServers() {
return bootstrapServers;
}
@@ -84,9 +73,9 @@ public class CanalClientConfig {
public static class CanalInstance {
private String instance;
private String instance; // 实例名
private List<AdapterGroup> adapterGroups;
private List<AdapterGroup> adapterGroups; // 适配器分组列表
public String getInstance() {
return instance;
@@ -109,7 +98,7 @@ public class CanalClientConfig {
public static class AdapterGroup {
private List<OuterAdapterConfig> outAdapters;
private List<OuterAdapterConfig> outAdapters; // 适配器列表
public List<OuterAdapterConfig> getOutAdapters() {
return outAdapters;
@@ -122,11 +111,11 @@ public class CanalClientConfig {
public static class MQTopic {
private String mqMode;
private String mqMode; // mq模式 kafka or rocketMQ
private String topic;
private String topic; // topic名
private List<Group> groups = new ArrayList<>();
private List<Group> groups = new ArrayList<>(); // 分组列表
public String getMqMode() {
return mqMode;
@@ -155,11 +144,11 @@ public class CanalClientConfig {
public static class Group {
private String groupId;
private String groupId; // group id
// private List<Adaptor> adapters = new ArrayList<>();
private List<OuterAdapterConfig> outAdapters;
private List<OuterAdapterConfig> outAdapters; // 适配器配置列表
public String getGroupId() {
return groupId;

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

@@ -1,21 +1,27 @@
package com.alibaba.otter.canal.client.adapter.support;
import com.alibaba.druid.pool.DruidDataSource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import com.alibaba.druid.pool.DruidDataSource;
/**
* 数据源配置
*
* @author rewerma @ 2018-10-20
* @version 1.0.0
*/
public class DatasourceConfig {
public final static Map<String, DruidDataSource> DATA_SOURCES = new ConcurrentHashMap<>();
public final static Map<String, DruidDataSource> DATA_SOURCES = new ConcurrentHashMap<>(); // key对应的数据源
private String driver = "com.mysql.jdbc.Driver";
private String url;
private String database;
private String type = "mysql";
private String username;
private String password;
private Integer maxActive = 3;
private String driver = "com.mysql.jdbc.Driver"; // 默认为mysql jdbc驱动
private String url; // jdbc url
private String database; // jdbc database
private String type = "mysql"; // 类型, 默认为mysql
private String username; // jdbc username
private String password; // jdbc password
private Integer maxActive = 3; // 连接池最大连接数,默认为3
public String getDriver() {
return driver;

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

@@ -7,24 +7,24 @@ import java.util.Map;
/**
* DML操作转换对象
*
* @author machengyuan 2018-8-19 下午11:30:49
* @author rewerma 2018-8-19 下午11:30:49
* @version 1.0.0
*/
public class Dml implements Serializable {
private static final long serialVersionUID = 2611556444074013268L;
private String destination;
private String database;
private String table;
private String type;
private String destination; // 对应canal的实例或者MQ的topic
private String database; // 数据库或schema
private String table; // 表名
private String type; // 类型: INSERT UPDATE DELETE
// binlog executeTime
private Long es;
private Long es; // 执行耗时
// dml build timeStamp
private Long ts;
private String sql;
private List<Map<String, Object>> data;
private List<Map<String, Object>> old;
private Long ts; // 同步时间
private String sql; // 执行的sql, dml sql为空
private List<Map<String, Object>> data; // 数据列表
private List<Map<String, Object>> old; // 旧数据列表, 用于update, size和data的size一一对应
public String getDestination() {
return destination;

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

@@ -2,14 +2,21 @@ package com.alibaba.otter.canal.client.adapter.support;
import java.io.Serializable;
/**
* ETL的结果对象
*
* @author rewerma @ 2018-10-20
* @version 1.0.0
*/
public class EtlResult implements Serializable {
private static final long serialVersionUID = 4250522736289866505L;
private boolean succeeded = false;
private boolean succeeded = false;
private String resultMessage;
private String resultMessage;
private String errorMessage;
private String errorMessage;
public boolean getSucceeded() {
return succeeded;
@@ -34,4 +41,4 @@ public class EtlResult implements Serializable {
public void setErrorMessage(String errorMessage) {
this.errorMessage = errorMessage;
}
}
}

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

@@ -1,22 +1,11 @@
package com.alibaba.otter.canal.client.adapter.support;
import java.io.BufferedReader;
import java.io.File;
import java.io.FilenameFilter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.*;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLClassLoader;
import java.nio.file.Paths;
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.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.regex.Pattern;
@@ -27,12 +16,13 @@ import org.slf4j.LoggerFactory;
/**
* SPI 类加载器
*
* @author machengyuan 2018-8-19 下午11:30:49
* @author rewerma 2018-8-19 下午11:30:49
* @version 1.0.0
*/
public class ExtensionLoader<T> {
private static final Logger logger = LoggerFactory.getLogger(ExtensionLoader.class);
private static final Logger logger = LoggerFactory
.getLogger(ExtensionLoader.class);
private static final String SERVICES_DIRECTORY = "META-INF/services/";
@@ -40,7 +30,8 @@ public class ExtensionLoader<T> {
private static final String DEFAULT_CLASSLOADER_POLICY = "internal";
private static final Pattern NAME_SEPARATOR = Pattern.compile("\\s*[,]+\\s*");
private static final Pattern NAME_SEPARATOR = Pattern
.compile("\\s*[,]+\\s*");
private static final ConcurrentMap<Class<?>, ExtensionLoader<?>> EXTENSION_LOADERS = new ConcurrentHashMap<>();
@@ -271,7 +262,8 @@ public class ExtensionLoader<T> {
return instance;
} catch (Throwable t) {
throw new IllegalStateException("Extension instance(name: " + name + ", class: " + type
+ ") could not be instantiated: " + t.getMessage(), t);
+ ") could not be instantiated: " + t.getMessage(),
t);
}
}
@@ -279,8 +271,8 @@ public class ExtensionLoader<T> {
if (type == null) throw new IllegalArgumentException("Extension type == null");
if (name == null) throw new IllegalArgumentException("Extension name == null");
Class<?> clazz = getExtensionClasses().get(name);
if (clazz == null) throw new IllegalStateException("No such extension \"" + name + "\" for " + type.getName()
+ "!");
if (clazz == null)
throw new IllegalStateException("No such extension \"" + name + "\" for " + type.getName() + "!");
return clazz;
}
@@ -342,8 +334,8 @@ public class ExtensionLoader<T> {
logger.info("extension classpath dir: " + dir);
File externalLibDir = new File(dir);
if (!externalLibDir.exists()) {
externalLibDir = new File(File.separator + this.getJarDirectoryPath() + File.separator + "canal_client"
+ File.separator + "lib");
externalLibDir = new File(
File.separator + this.getJarDirectoryPath() + File.separator + "canal_client" + File.separator + "lib");
}
if (externalLibDir.exists()) {
File[] files = externalLibDir.listFiles(new FilenameFilter() {
@@ -495,12 +487,10 @@ public class ExtensionLoader<T> {
// Class.forName(line, true,
// classLoader);
if (!type.isAssignableFrom(clazz)) {
throw new IllegalStateException("Error when load extension class(interface: "
+ type
+ ", class line: "
+ clazz.getName()
+ "), class "
+ clazz.getName()
throw new IllegalStateException(
"Error when load extension class(interface: " + type
+ ", class line: " + clazz.getName()
+ "), class " + clazz.getName()
+ "is not subtype of interface.");
} else {
try {
@@ -518,9 +508,9 @@ public class ExtensionLoader<T> {
extensionClasses.put(n, clazz);
} else if (c != clazz) {
cachedNames.remove(clazz);
throw new IllegalStateException("Duplicate extension "
+ type.getName()
+ " name " + n + " on "
throw new IllegalStateException(
"Duplicate extension " + type.getName() + " name "
+ n + " on "
+ c.getName() + " and "
+ clazz.getName());
}
@@ -530,12 +520,9 @@ public class ExtensionLoader<T> {
}
}
} catch (Throwable t) {
IllegalStateException e = new IllegalStateException("Failed to load extension class(interface: "
+ type
+ ", class line: "
+ line
+ ") in "
+ url
IllegalStateException e = new IllegalStateException(
"Failed to load extension class(interface: " + type + ", class line: "
+ line + ") in " + url
+ ", cause: "
+ t.getMessage(),
t);
@@ -550,13 +537,15 @@ public class ExtensionLoader<T> {
}
} catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + type + ", class file: " + url
+ ") in " + url, t);
+ ") in " + url,
t);
}
} // end of while urls
}
} catch (Throwable t) {
logger.error("Exception when load extension class(interface: " + type + ", description file: " + fileName
+ ").", t);
logger.error(
"Exception when load extension class(interface: " + type + ", description file: " + fileName + ").",
t);
}
}

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

@@ -11,7 +11,7 @@ import org.slf4j.LoggerFactory;
/**
* 类型转换工具类
*
* @author machengyuan 2018-8-19 下午06:14:23
* @author rewerma 2018-8-19 下午06:14:23
* @version 1.0.0
*/
public class JdbcTypeUtil {
@@ -30,7 +30,7 @@ public class JdbcTypeUtil {
switch (jdbcType) {
case Types.BIT:
case Types.BOOLEAN:
// return Boolean.class;
// return Boolean.class;
case Types.TINYINT:
return Byte.TYPE;
case Types.SMALLINT:

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

@@ -9,7 +9,7 @@ import com.alibaba.otter.canal.protocol.Message;
/**
* Message对象解析工具类
*
* @author machengyuan 2018-8-19 下午06:14:23
* @author rewerma 2018-8-19 下午06:14:23
* @version 1.0.0
*/
public class MessageUtil {

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

@@ -5,7 +5,7 @@ import java.util.Map;
/**
* 外部适配器配置信息类
*
* @author machengyuan 2018-8-18 下午10:15:12
* @author rewerma 2018-8-18 下午10:15:12
* @version 1.0.0
*/
public class OuterAdapterConfig {

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

@@ -3,9 +3,15 @@ package com.alibaba.otter.canal.client.adapter.support;
import java.io.Serializable;
import java.util.Date;
/**
* 用于rest的结果返回对象
*
* @author rewerma @ 2018-10-20
* @version 1.0.0
*/
public class Result implements Serializable {
public Integer code = 20000;
public Integer code = 20000;
public Object data;
public String message;
public Date sysTime;

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

@@ -6,6 +6,12 @@ import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
/**
* SPI装载器注解
*
* @author rewerma @ 2018-10-20
* @version 1.0.0
*/
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ ElementType.TYPE })