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
2019-02-13 15:47:48 +08:00
parent 83b760494d
commit 0ef21e4155
3 changed files with 15 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
View File

@@ -144,7 +144,9 @@ public class ESAdapter implements OuterAdapter {
return; return;
} }
for (Dml dml : dmls) { for (Dml dml : dmls) {
sync(dml); if (!dml.getIsDdl()) {
sync(dml);
}
} }
esSyncService.commit(); // 批次统一提交 esSyncService.commit(); // 批次统一提交
} }

9
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
View File

@@ -71,7 +71,12 @@ public class ESSyncService {
dml.getDestination()); dml.getDestination());
} }
if (logger.isDebugEnabled()) { if (logger.isDebugEnabled()) {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue)); StringBuilder configIndexes = new StringBuilder();
esSyncConfigs
.forEach(esSyncConfig -> configIndexes.append(esSyncConfig.getEsMapping().get_index()).append(" "));
logger.debug("DML: {} \nEffect indexes: {}",
JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue),
configIndexes.toString());
} }
} }
} }
@@ -92,6 +97,8 @@ public class ESSyncService {
update(config, dml); update(config, dml);
} else if (type != null && type.equalsIgnoreCase("DELETE")) { } else if (type != null && type.equalsIgnoreCase("DELETE")) {
delete(config, dml); delete(config, dml);
} else {
return;
} }
if (logger.isTraceEnabled()) { if (logger.isTraceEnabled()) {

8
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
View File

@@ -65,8 +65,8 @@ public class ESTemplate {
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
if (mapping.isUpsert()) { if (mapping.isUpsert()) {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setDoc(esFieldData) .setDoc(esFieldData)
.setDocAsUpsert(true)); .setDocAsUpsert(true));
} else { } else {
getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
.setSource(esFieldData)); .setSource(esFieldData));
@@ -200,7 +200,7 @@ public class ESTemplate {
* 提交批次 * 提交批次
*/ */
public void commit() { public void commit() {
if (getBulk().numberOfActions() >= 0) { if (getBulk().numberOfActions() > 0) {
BulkResponse response = getBulk().execute().actionGet(); BulkResponse response = getBulk().execute().actionGet();
if (response.hasFailures()) { if (response.hasFailures()) {
for (BulkItemResponse itemResponse : response.getItems()) { for (BulkItemResponse itemResponse : response.getItems()) {
@@ -369,7 +369,7 @@ public class ESTemplate {
resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName); resultIdVal = getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName);
} }
if (dmlOld.get(columnName) != null && !mapping.getSkips().contains(fieldItem.getFieldName())) { if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) {
esFieldData.put(fieldItem.getFieldName(), esFieldData.put(fieldItem.getFieldName(),
getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName));
} }