Skip to content

Commit a39fc2c

Browse files
sasousasou
committed
Merge branch 'develop'
2 parents 427af6c + bd9782c commit a39fc2c

11 files changed

Lines changed: 58 additions & 80 deletions

File tree

‎README.md‎

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -60,7 +60,7 @@ search build
6060
Topic规则：数据库的每个表有单独的topic，如数据库admin的user表，对应的kafka主题名为：sync_admin_user  
6161
Topic数据字段：
6262

63-
插入数据：
63+
插入数据同步格式：
6464
{
6565
"head": {
6666
"binlog_pos": 53036,
@@ -82,7 +82,7 @@ Topic数据字段：
8282
]
8383
}
8484

85-
修改数据：
85+
修改数据同步格式：
8686
{
8787
"head": {
8888
"binlog_pos": 53036,
@@ -115,7 +115,7 @@ Topic数据字段：
115115
]
116116
}
117117

118-
删除数据：
118+
删除数据同步格式：
119119
{
120120
"head": {
121121
"binlog_pos": 53036,
@@ -158,4 +158,13 @@ List规则：数据库的每个表有单独的list，如数据库admin的user表
158158

159159
**Elasticsearch**
160160

161-
规则：同步index = sync，如数据库admin的user表，对应的Elasticsearch type名为：admin_user  
161+
规则：数据库的每个表有单独的Elasticsearch index，如数据库admin的user表，对应的es index名为：sync_admin_user, index type 为default;
162+
163+
Elasticsearch同步数据的head中有id字段；
164+
165+
Mysql 同步到 Elasticsearch注意事项：
166+
167+
1、表需要有一个唯一id主键；
168+
2、表时间字段datetime会转为es的时间字段，其他字段对应es的文本类型；
169+
3、主键、时间字段禁止修改，其他字段尽量提前规划好；
170+

‎bin/SysConfig.properties‎

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -2,24 +2,19 @@
22
system.debug=1
33

44
#canal
5-
canal.ip=127.0.0.1
5+
canal.ip=10.5.3.70
66
canal.port=11111
7-
canal.destination=sdsw,epos,es
7+
canal.destination=es,redis
88
canal.username=
99
canal.password=
1010
canal.filter=
1111

1212
#redis
13-
sdsw.target_type=redis
14-
sdsw.target_ip=192.168.27.101
15-
sdsw.target_port=
16-
17-
#kafka
18-
epos.target_type=kafka
19-
epos.target_ip=127.0.0.1
20-
epos.target_port=
13+
redis.target_type=redis
14+
redis.target_ip=10.5.3.67
15+
redis.target_port=
2116

2217
#elasticsearch
2318
es.target_type=elasticsearch
24-
es.target_ip=127.0.0.1
19+
es.target_ip=10.5.3.72
2520
es.target_port=

‎bin/syncClient.jar‎

-118 Bytes
Binary file not shown.

‎src/SysConfig.properties‎

Lines changed: 8 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -2,24 +2,19 @@
22
system.debug=1
33

44
#canal
5-
canal.ip=127.0.0.1
5+
canal.ip=10.5.3.70
66
canal.port=11111
7-
canal.destination=sdsw
7+
canal.destination=es,redis
88
canal.username=
99
canal.password=
1010
canal.filter=
1111

1212
#redis
13-
test.target_type=redis
14-
test.target_ip=127.0.0.1
15-
test.target_port=
16-
17-
#kafka
18-
epos.target_type=kafka
19-
epos.target_ip=127.0.0.1
20-
epos.target_port=
13+
redis.target_type=redis
14+
redis.target_ip=10.5.3.67
15+
redis.target_port=
2116

2217
#elasticsearch
23-
sdsw.target_type=elasticsearch
24-
sdsw.target_ip=10.5.3.66
25-
sdsw.target_port=
18+
es.target_type=elasticsearch
19+
es.target_ip=10.5.3.72
20+
es.target_port=

‎src/src/com/sync/common/EsApi.java‎

Lines changed: 3 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -44,16 +44,14 @@ public boolean sync(String index, String content) throws Exception {
4444
Map<String,Object> data = jsonToMap(content);
4545
Map<String,Object> head = jsonToMap((String) data.get("head").toString());
4646
String type = (String) head.get("type").toString();
47-
String db = (String) head.get("db").toString();
48-
String table = (String) head.get("table").toString();
4947
String id = (String) head.get("id").toString();
5048
String text = "";
5149
switch(type) {
5250
case "INSERT":
5351
text = (String) data.get("after").toString();
5452
if (!"".equals(text)) {
5553
try {
56-
return insert("sync" + "-" + db + "-"+ table, "default", id, text);
54+
return insert(index, "default", id, text);
5755
} catch (Exception e) {
5856
throw new Exception("elasticsearch insert fail", e);
5957
}
@@ -63,7 +61,7 @@ public boolean sync(String index, String content) throws Exception {
6361
text = (String) data.get("after").toString();
6462
if (!"".equals(id)) {
6563
try {
66-
return update("sync" + "-" + db + "-"+ table, "default", id, text);
64+
return update(index, "default", id, text);
6765
} catch (Exception e) {
6866
throw new Exception("elasticsearch update fail", e);
6967
}
@@ -72,7 +70,7 @@ public boolean sync(String index, String content) throws Exception {
7270
case "DELETE":
7371
if (!"".equals(id)) {
7472
try {
75-
return delete("sync" + "-" + db + "-"+ table, "default", id);
73+
return delete(index, "default", id);
7674
} catch (Exception e) {
7775

7876
}
@@ -101,9 +99,6 @@ public static Map<String,Object> jsonToMap(String jsonObj) {
10199
* @throws Exception
102100
*/
103101
public boolean insert(String index, String type, String id, String content) throws Exception {
104-
if (!index("sync-sdsw-sys_log")) {
105-
System.out.println(setMappings("sync-sdsw-sys_log"));
106-
}
107102
Map<String, String> params = Collections.emptyMap();
108103
HttpEntity entity = new NStringEntity(content, ContentType.APPLICATION_JSON);
109104
Response response = rs.performRequest("PUT", "/" + index + "/" + type + "/" + id, params, entity);
@@ -119,9 +114,6 @@ public boolean insert(String index, String type, String id, String content) thro
119114
* @throws Exception
120115
*/
121116
public boolean update(String index, String type, String id, String content) throws Exception {
122-
if (!index("sync-sdsw-sys_log")) {
123-
System.out.println(setMappings("sync-sdsw-sys_log"));
124-
}
125117
Map<String, String> params = Collections.emptyMap();
126118
HttpEntity entity = new NStringEntity(content, ContentType.APPLICATION_JSON);
127119
Response response = rs.performRequest("PUT", "/" + index + "/" + type + "/" + id, params, entity);

‎src/src/com/sync/common/GetProperties.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
import com.sync.common.ReadProperties;
88

99
/**
10-
* config
10+
* GetProperties
1111
*
1212
* @author sasou <admin@php-gene.com> web:http://www.php-gene.com/
1313
* @version 1.0.0

‎src/src/com/sync/common/RedisApi.java‎

Lines changed: 16 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,7 @@
99
import redis.clients.jedis.exceptions.JedisConnectionException;
1010

1111
/**
12-
* config
12+
* RedisApi
1313
*
1414
* @author sasou <admin@php-gene.com> web:http://www.php-gene.com/
1515
* @version 1.0.0
@@ -22,28 +22,18 @@ public class RedisApi {
2222
public RedisApi(String name) {
2323
canal_destination = name;
2424
JedisPoolConfig config = new JedisPoolConfig();
25-
// 控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；
26-
// 如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted(耗尽)。
2725
config.setMaxTotal(1000);
28-
// 控制一个pool最多有多少个状态为idle(空闲的)的jedis实例。
2926
config.setMaxIdle(50);
30-
// 表示当borrow(引入)一个jedis实例时，最大的等待时间，如果超过等待时间，则直接抛出JedisConnectionException；
3127
config.setMaxWaitMillis(1000 * 10);
32-
33-
// 在获取连接的时候检查有效性, 默认false
3428
config.setTestOnBorrow(false);
35-
// 在空闲时检查有效性, 默认false
3629
config.setTestWhileIdle(false);
37-
38-
// 连接耗尽时是否阻塞, false报异常,ture阻塞直到超时, 默认true
3930
config.setBlockWhenExhausted(true);
40-
4131
config.setMinEvictableIdleTimeMillis(300000);
4232
pool = new JedisPool(config, GetProperties.target.get(canal_destination).ip, GetProperties.target.get(canal_destination).port, 1000 * 10);
4333
}
4434

4535
/**
46-
* 返还到连接池
36+
* return Resource to pool
4737
*
4838
* @param pool
4939
* @param redis
@@ -55,7 +45,7 @@ public void returnResource(JedisPool pool, Jedis redis) {
5545
}
5646

5747
/**
58-
* 获取数据
48+
* get data
5949
*
6050
* @param key
6151
* @return
@@ -74,7 +64,7 @@ public String get(String key) throws Exception {
7464
}
7565

7666
/**
77-
* 获取Set数据
67+
* zrange data
7868
*
7969
* @param key
8070
* @return
@@ -94,7 +84,7 @@ public Set zrange(String key) throws Exception {
9484
}
9585

9686
/**
97-
* 获取数据
87+
* lrange data
9888
*
9989
* @param key
10090
* @return
@@ -114,7 +104,7 @@ public List lrange(String key) throws Exception {
114104
}
115105

116106
/**
117-
* 写String数据
107+
* set string
118108
*
119109
* @param key
120110
* @return
@@ -132,7 +122,7 @@ public void set(String key, String value) throws Exception {
132122
}
133123

134124
/**
135-
* 写set数据
125+
* set data
136126
*
137127
* @param key
138128
* @return
@@ -150,7 +140,7 @@ public void zadd(String key, String member) throws Exception {
150140
}
151141

152142
/**
153-
* 写List数据 左添加
143+
* push list in left
154144
*
155145
* @param key
156146
* @return
@@ -167,7 +157,7 @@ public void lpush(String key, String member) throws Exception {
167157
}
168158

169159
/**
170-
* 写List数据 右添加
160+
* push list in right
171161
*
172162
* @param key
173163
* @return
@@ -184,7 +174,7 @@ public void rpush(String key, String member) throws Exception {
184174
}
185175

186176
/**
187-
* 写数据
177+
* exists
188178
*
189179
* @param key
190180
* @return
@@ -204,7 +194,7 @@ public boolean exists(String key) throws Exception {
204194
}
205195

206196
/**
207-
* 写数据
197+
* del
208198
*
209199
* @param key
210200
* @return
@@ -221,7 +211,7 @@ public void del(String key) throws Exception {
221211
}
222212

223213
/**
224-
* 删除List元素
214+
* lrem
225215
*
226216
* @param key
227217
* @return
@@ -238,7 +228,7 @@ public void lrem(String key, String member) throws Exception {
238228
}
239229

240230
/**
241-
* 删除set元素
231+
* zrem
242232
*
243233
* @param key
244234
* @return
@@ -255,11 +245,10 @@ public void zrem(String key, String member) throws Exception {
255245
}
256246

257247
/**
258-
* 设置过期时间
248+
* expire
259249
*
260250
* @param key
261251
* @param num
262-
* 过期时间 分钟
263252
*/
264253
public void expire(String key, int num) throws Exception {
265254
Jedis jedis = null;
@@ -273,7 +262,7 @@ public void expire(String key, int num) throws Exception {
273262
}
274263

275264
/**
276-
* 执行+1操作
265+
* incr 1
277266
*
278267
* @param key
279268
*/
@@ -289,10 +278,9 @@ public void incr(String key) throws Exception {
289278
}
290279

291280
/**
292-
* 清空
281+
* clear
293282
*
294283
* @param num
295-
* 过期时间 分钟
296284
*/
297285
public void clear() throws Exception {
298286
Jedis jedis = null;

‎src/src/com/sync/common/WriteLog.java‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
package com.sync.common;
22

33
/**
4-
* TargetData
4+
* WriteLog
55
*
66
* @author sasou <admin@php-gene.com> web:http://www.php-gene.com/
77
* @version 1.0.0
@@ -11,10 +11,9 @@
1111
import java.util.Calendar;
1212

1313
/**
14-
* 写日志 写logString字符串到./log目录下的文件中
14+
* write logString
1515
*
1616
* @param logString
17-
* 日志字符串
1817
* @author tower
1918
*/
2019
public class WriteLog {
@@ -25,7 +24,7 @@ public static void write(String type, String logString) {
2524
base = System.getProperty("user.dir");
2625
}
2726

28-
String current = base + "\\logs\\";
27+
String current = base + "/logs/";
2928
try {
3029
String logFilePathName = null;
3130
Calendar cd = Calendar.getInstance();
@@ -46,7 +45,7 @@ public static void write(String type, String logString) {
4645

4746
FileOutputStream fos = new FileOutputStream(logFilePathName, true);
4847
String time = "[" + year + "-" + month + "-" + day + " " + hour + ":" + min + ":" + sec + "] ";
49-
String content = time + logString + "\n\n";
48+
String content = time + logString + "\r\n";
5049
fos.write(content.getBytes());
5150
fos.flush();
5251
fos.close();
@@ -59,7 +58,7 @@ public static void write(String type, String logString) {
5958
}
6059

6160
/**
62-
* 整数i小于10则前面补0
61+
* add 0
6362
*
6463
* @param i
6564
* @return

0 commit comments

Comments
 (0)