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

es 批量提交修改

Browse Source
This commit is contained in:
mcy
2019-02-22 16:56:58 +08:00
parent 01dc8ce2b8
commit 17efc2ada0
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -72,6 +72,7 @@ public class ESTemplate {
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));
} }
commitBulk();
} else { } else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type()) .setTypes(mapping.get_type())
@@ -81,9 +82,10 @@ public class ESTemplate {
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
.setDoc(esFieldData)); .setDoc(esFieldData));
commitBulk();
} }
} }
commitBulk();
} }
/** /**
@@ -149,6 +151,7 @@ public class ESTemplate {
public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) { public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
if (mapping.get_id() != null) { if (mapping.get_id() != null) {
getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString()));
commitBulk();
} else { } else {
SearchResponse response = transportClient.prepareSearch(mapping.get_index()) SearchResponse response = transportClient.prepareSearch(mapping.get_index())
.setTypes(mapping.get_type()) .setTypes(mapping.get_type())
@@ -158,9 +161,10 @@ public class ESTemplate {
for (SearchHit hit : response.getHits()) { for (SearchHit hit : response.getHits()) {
getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
.setDoc(esFieldData)); .setDoc(esFieldData));
commitBulk();
} }
} }
commitBulk();
} }
/** /**
@@ -186,7 +190,7 @@ public class ESTemplate {
} }
/** /**
* 如果大于批量则提交批次 * 如果大于批量数则提交批次
*/ */
private void commitBulk() { private void commitBulk() {
if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) { if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {