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

fixed canal-admin vue

Browse Source
This commit is contained in:
agapple
2019-08-24 16:54:27 +08:00
parent 983ff5fc73
commit e376a129ed
36 changed files with 146 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,24 +1,26 @@
package com.alibaba.otter.canal.client.adapter.support;
import com.alibaba.druid.pool.DruidDataSource;
import com.google.common.base.Joiner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import javax.sql.DataSource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.pool.DruidDataSource;
import com.google.common.base.Joiner;
public abstract class AbstractEtlService {
protected Logger logger = LoggerFactory.getLogger(this.getClass());
protected Logger logger = LoggerFactory.getLogger(this.getClass());
private String type;
private AdapterConfig config;
private final long CNT_PER_TASK = 10000L;
private final long CNT_PER_TASK = 10000L;
public AbstractEtlService(String type, AdapterConfig config){
this.type = type;
@@ -43,7 +45,6 @@ public abstract class AbstractEtlService {
// 拼接条件
if (config.getMapping().getEtlCondition() != null && params != null) {
String etlCondition = config.getMapping().getEtlCondition();
int size = params.size();
for (String param : params) {
etlCondition = etlCondition.replace("{}", "?");
values.add(param);
@@ -87,8 +88,12 @@ public abstract class AbstractEtlService {
for (long i = 0; i < workerCnt; i++) {
offset = size * i;
String sqlFinal = sql + " LIMIT " + offset + "," + size;
Future<Boolean> future = executor.submit(
() -> executeSqlImport(dataSource, sqlFinal, values, config.getMapping(), impCount, errMsg));
Future<Boolean> future = executor.submit(() -> executeSqlImport(dataSource,
sqlFinal,
values,
config.getMapping(),
impCount,
errMsg));
futures.add(future);
}

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

@@ -6,7 +6,6 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import com.alibaba.fastjson.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -152,8 +151,9 @@ public abstract class AbstractCanalAdapterWorker {
if (flatMessage) {
// batch write
writeOut((List<FlatMessage>) messages);
//FIXME xxx
messages.forEach((message -> System.out.println(JSON.toJSONString(message))));
// FIXME xxx
// messages.forEach((message ->
// System.out.println(JSON.toJSONString(message))));
} else {
for (final Object message : messages) {
writeOut((Message) message);

16
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/RdbAdapter.java
View File

@@ -23,7 +23,11 @@ import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService;
import com.alibaba.otter.canal.client.adapter.rdb.service.RdbMirrorDbSyncService;
import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService;
import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
import com.alibaba.otter.canal.client.adapter.support.*;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.client.adapter.support.Util;
/**
* RDB适配器实现类
@@ -73,8 +77,8 @@ public class RdbAdapter implements OuterAdapter {
// 过滤不匹配的key的配置
rdbMappingTmp.forEach((key, mappingConfig) -> {
if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
|| (mappingConfig.getOuterAdapterKey() != null
&& mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
|| (mappingConfig.getOuterAdapterKey() != null && mappingConfig.getOuterAdapterKey()
.equalsIgnoreCase(configuration.getKey()))) {
rdbMapping.put(key, mappingConfig);
}
});
@@ -134,8 +138,8 @@ public class RdbAdapter implements OuterAdapter {
String threads = properties.get("threads");
// String commitSize = properties.get("commitSize");
boolean skipDupException = BooleanUtils
.toBoolean(configuration.getProperties().getOrDefault("skipDupException", "true"));
boolean skipDupException = BooleanUtils.toBoolean(configuration.getProperties()
.getOrDefault("skipDupException", "true"));
rdbSyncService = new RdbSyncService(dataSource,
threads != null ? Integer.valueOf(threads) : null,
skipDupException);
@@ -181,7 +185,7 @@ public class RdbAdapter implements OuterAdapter {
MappingConfig config = rdbMapping.get(task);
RdbEtlService rdbEtlService = new RdbEtlService(dataSource, config);
if (config != null) {
return rdbEtlService.importData(params);
return rdbEtlService.importData(params);
} else {
StringBuilder resultMsg = new StringBuilder();
boolean resSucc = true;

5
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

@@ -105,6 +105,11 @@ public class RdbSyncService {
List<Future<Boolean>> futures = new ArrayList<>();
for (int i = 0; i < threads; i++) {
int j = i;
if (dmlsPartition[j].isEmpty()) {
// bypass
continue;
}
futures.add(executorThreads[i].submit(() -> {
try {
dmlsPartition[j].forEach(syncItem -> sync(batchExecutors[j],