增加输出数据的集合或者队列的多种名称规则；

sasou · Jul 25, 2018 · 4180387 · 4180387
1 parent fb4abed
commit 4180387
Show file tree

Hide file tree

Showing 9 changed files with 60 additions and 6 deletions.
diff --git a/bin/SysConfig.properties b/bin/SysConfig.properties
@@ -22,4 +22,5 @@ es.target_port=
 #httpmq
 ztbiao.target_type=httpmq
 ztbiao.target_ip=10.5.5.15
-ztbiao.target_port=1218
+ztbiao.target_port=1218
+ztbiao.target_deep=2
diff --git a/bin/syncClient.jar b/bin/syncClient.jar
diff --git a/src/src/com/sync/common/GetProperties.java b/src/src/com/sync/common/GetProperties.java
@@ -84,6 +84,10 @@ public GetProperties() {
 					if (!"".equals(tmp)) {
 						target_tmp.setPort(Integer.parseInt(tmp));
 					}
+					tmp = String.valueOf(p.get(canal.destination[i] + ".target_deep"));
+					if (!"".equals(tmp)) {
+						target_tmp.setDeep(Integer.parseInt(tmp));
+					}
 					target.put(canal.destination[i], target_tmp);
 				}
 			}

diff --git a/src/src/com/sync/common/TargetData.java b/src/src/com/sync/common/TargetData.java
@@ -10,6 +10,7 @@ public final class TargetData {
 	public String type = "";
 	public String ip = "";
 	public int port = 0;
+	public int deep = 1;
 
 	/**
 	 * @return the type
@@ -55,4 +56,20 @@ public int getPort() {
 	public void setPort(int port) {
 		this.port = port;
 	}
+
+	/**
+	 * @return the deep
+	 */
+	public int getDeep() {
+		return deep;
+	}
+
+	/**
+	 * @param deep
+	 *            the deep to set
+	 */
+	public void setDeep(int deep) {
+		this.deep = deep;
+	}
+
 }
diff --git a/src/src/com/sync/common/Tool.java b/src/src/com/sync/common/Tool.java
@@ -0,0 +1,28 @@
+package com.sync.common;
+
+public class Tool {
+
+	public static String makeTargetName(String canal_destination, String db, String table) {
+	    int type = GetProperties.target.get(canal_destination).deep;
+	    String ret = null;
+		switch(type) {
+		case 1:
+			ret = "sync_" + canal_destination + "_" + db + "_" + table;
+			break;
+		case 2:
+			ret = "sync_" + canal_destination + "_" + db;
+			break;
+		case 3:
+			ret = "sync_" + canal_destination;
+			break;
+		case 4:
+			ret = "sync_" + "_" + db + "_" + table;
+			break;
+		default:
+			ret = "sync_" + canal_destination + "_" + db + "_" + table;
+			break;
+		}
+		return ret;
+	}
+
+}
diff --git a/src/src/com/sync/process/ElasticSearch.java b/src/src/com/sync/process/ElasticSearch.java
@@ -20,6 +20,7 @@
 import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
 import com.sync.common.EsApi;
 import com.sync.common.GetProperties;
+import com.sync.common.Tool;
 import com.sync.common.WriteLog;
 import com.alibaba.fastjson.JSON;
 
@@ -117,7 +118,7 @@ private boolean syncEntry(List<Entry> entrys) {
 			head.put("type", eventType);
 			data.put("head", head);
 
-			topic = "sync_" + entry.getHeader().getSchemaName() + "_" + entry.getHeader().getTableName();
+			topic = Tool.makeTargetName(canal_destination, entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
 			no = (int) entry.getHeader().getLogfileOffset();
 			for (RowData rowData : rowChage.getRowDatasList()) {
 				if (eventType == EventType.DELETE) {

diff --git a/src/src/com/sync/process/Httpmq.java b/src/src/com/sync/process/Httpmq.java
@@ -16,6 +16,7 @@
 import com.sync.common.GetProperties;
 import com.sync.common.HttpmqApi;
 import com.sync.common.RedisApi;
+import com.sync.common.Tool;
 import com.sync.common.WriteLog;
 import com.alibaba.fastjson.JSON;
 
@@ -110,7 +111,7 @@ private boolean syncEntry(List<Entry> entrys) {
 			head.put("table", entry.getHeader().getTableName());
 			head.put("type", eventType);
 			data.put("head", head);
-			topic = "sync_" + canal_destination + "_" + entry.getHeader().getSchemaName() + "_" + entry.getHeader().getTableName();
+			topic = Tool.makeTargetName(canal_destination, entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
 			no = (int) entry.getHeader().getLogfileOffset();
 			for (RowData rowData : rowChage.getRowDatasList()) {
 				if (eventType == EventType.DELETE) {
@@ -139,7 +140,7 @@ private boolean syncEntry(List<Entry> entrys) {
 		}
 		return ret;
 	}
-
+	
 	private Map<String, Object> makeColumn(List<Column> columns) {
 		Map<String, Object> one = new HashMap<String, Object>();
 		for (Column column : columns) {

diff --git a/src/src/com/sync/process/Kafka.java b/src/src/com/sync/process/Kafka.java
@@ -20,6 +20,7 @@
 import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
 import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
 import com.sync.common.GetProperties;
+import com.sync.common.Tool;
 import com.sync.common.WriteLog;
 import com.alibaba.fastjson.JSON;
 
@@ -126,7 +127,7 @@ private boolean syncEntry(List<Entry> entrys) {
 			head.put("table", entry.getHeader().getTableName());
 			head.put("type", eventType);
 			data.put("head", head);
-			topic = "sync_" + entry.getHeader().getSchemaName() + "_" + entry.getHeader().getTableName();
+			topic = Tool.makeTargetName(canal_destination, entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
 			no = (int) entry.getHeader().getLogfileOffset();
 			for (RowData rowData : rowChage.getRowDatasList()) {
 				if (eventType == EventType.DELETE) {

diff --git a/src/src/com/sync/process/Redis.java b/src/src/com/sync/process/Redis.java
@@ -15,6 +15,7 @@
 import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
 import com.sync.common.GetProperties;
 import com.sync.common.RedisApi;
+import com.sync.common.Tool;
 import com.sync.common.WriteLog;
 import com.alibaba.fastjson.JSON;
 
@@ -109,7 +110,7 @@ private boolean syncEntry(List<Entry> entrys) {
 			head.put("table", entry.getHeader().getTableName());
 			head.put("type", eventType);
 			data.put("head", head);
-			topic = "sync_" + entry.getHeader().getSchemaName() + "_" + entry.getHeader().getTableName();
+			topic = Tool.makeTargetName(canal_destination, entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
 			no = (int) entry.getHeader().getLogfileOffset();
 			for (RowData rowData : rowChage.getRowDatasList()) {
 				if (eventType == EventType.DELETE) {