From fdd07635673bc0e2a3219ff1a2825fb7d352a75e Mon Sep 17 00:00:00 2001 From: rewerma Date: Wed, 21 Aug 2019 12:44:57 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9Ees=20adapter=20rest=20client?= =?UTF-8?q?=E5=92=8C=E8=AE=A4=E8=AF=81=E6=9C=BA=E5=88=B6=20(#2046)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 新增es adapter rest client和认证机制 * 修改getMapping位置 * 代码调整 * 代码调整 * 整理代码 * fix #2048 --- client-adapter/elasticsearch/pom.xml | 14 +- .../canal/client/adapter/es/ESAdapter.java | 51 +- .../adapter/es/service/ESEtlService.java | 77 +-- .../adapter/es/support/ESConnection.java | 468 ++++++++++++++++ .../client/adapter/es/support/ESTemplate.java | 189 +++---- .../client/RequestConvertersExt.java | 30 ++ .../client/RestHighLevelClientExt.java | 29 + .../es/test/sync/LabelSyncJoinSub2Test.java | 16 +- .../es/test/sync/LabelSyncJoinSubTest.java | 16 +- .../es/test/sync/RoleSyncJoinOne2Test.java | 11 +- .../es/test/sync/RoleSyncJoinOneTest.java | 11 +- .../es/test/sync/UserSyncJoinOneTest.java | 11 +- .../es/test/sync/UserSyncSingleTest.java | 16 +- .../monitor/remote/DbRemoteConfigLoader.java | 502 +++++++++--------- .../src/main/resources/application.yml | 11 +- .../otter/canal/kafka/CanalKafkaProducer.java | 1 + 16 files changed, 1012 insertions(+), 441 deletions(-) create mode 100644 client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESConnection.java create mode 100644 client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RequestConvertersExt.java create mode 100644 client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RestHighLevelClientExt.java diff --git a/client-adapter/elasticsearch/pom.xml b/client-adapter/elasticsearch/pom.xml index 79136dad..836ef306 100644 --- a/client-adapter/elasticsearch/pom.xml +++ b/client-adapter/elasticsearch/pom.xml @@ -26,12 +26,22 @@ org.elasticsearch elasticsearch - 6.2.3 + 6.4.3 org.elasticsearch.client transport - 6.2.3 + 6.4.3 + + + org.elasticsearch.client + elasticsearch-rest-client + 6.4.3 + + + org.elasticsearch.client + elasticsearch-rest-high-level-client + 6.4.3 diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java index 16f9e6cb..1ae8c638 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java @@ -1,6 +1,5 @@ package com.alibaba.otter.canal.client.adapter.es; -import java.net.InetAddress; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -13,10 +12,6 @@ import javax.sql.DataSource; import org.apache.commons.lang.StringUtils; import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.client.transport.TransportClient; -import org.elasticsearch.common.settings.Settings; -import org.elasticsearch.common.transport.TransportAddress; -import org.elasticsearch.transport.client.PreBuiltTransportClient; import com.alibaba.druid.pool.DruidDataSource; import com.alibaba.otter.canal.client.adapter.OuterAdapter; @@ -28,6 +23,7 @@ import com.alibaba.otter.canal.client.adapter.es.config.SqlParser; import com.alibaba.otter.canal.client.adapter.es.monitor.ESConfigMonitor; import com.alibaba.otter.canal.client.adapter.es.service.ESEtlService; import com.alibaba.otter.canal.client.adapter.es.service.ESSyncService; +import com.alibaba.otter.canal.client.adapter.es.support.ESConnection; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.support.*; @@ -43,7 +39,7 @@ public class ESAdapter implements OuterAdapter { private Map esSyncConfig = new ConcurrentHashMap<>(); // 文件名对应配置 private Map> dbTableEsSyncConfig = new ConcurrentHashMap<>(); // schema-table对应配置 - private TransportClient transportClient; + private ESConnection esConnection; private ESSyncService esSyncService; @@ -51,10 +47,6 @@ public class ESAdapter implements OuterAdapter { private Properties envProperties; - public TransportClient getTransportClient() { - return transportClient; - } - public ESSyncService getEsSyncService() { return esSyncService; } @@ -67,6 +59,14 @@ public class ESAdapter implements OuterAdapter { return dbTableEsSyncConfig; } + public ESConnection getEsConnection() { + return esConnection; + } + + public void setEsConnection(ESConnection esConnection) { + this.esConnection = esConnection; + } + @Override public void init(OuterAdapterConfig configuration, Properties envProperties) { try { @@ -119,17 +119,16 @@ public class ESAdapter implements OuterAdapter { } Map properties = configuration.getProperties(); - Settings.Builder settingBuilder = Settings.builder(); - properties.forEach(settingBuilder::put); - Settings settings = settingBuilder.build(); - transportClient = new PreBuiltTransportClient(settings); + String[] hostArray = configuration.getHosts().split(","); - for (String host : hostArray) { - int i = host.indexOf(":"); - transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)), - Integer.parseInt(host.substring(i + 1)))); + String mode = properties.get("mode"); + if ("rest".equalsIgnoreCase(mode) || "http".equalsIgnoreCase(mode)) { + esConnection = new ESConnection(hostArray, properties, ESConnection.ESClientMode.REST); + } else { + esConnection = new ESConnection(hostArray, properties, ESConnection.ESClientMode.TRANSPORT); } - ESTemplate esTemplate = new ESTemplate(transportClient); + + ESTemplate esTemplate = new ESTemplate(esConnection); esSyncService = new ESSyncService(esTemplate); esConfigMonitor = new ESConfigMonitor(); @@ -177,7 +176,7 @@ public class ESAdapter implements OuterAdapter { ESSyncConfig config = esSyncConfig.get(task); if (config != null) { DataSource dataSource = DatasourceConfig.DATA_SOURCES.get(config.getDataSourceKey()); - ESEtlService esEtlService = new ESEtlService(transportClient, config); + ESEtlService esEtlService = new ESEtlService(esConnection, config); if (dataSource != null) { return esEtlService.importData(params); } else { @@ -191,7 +190,7 @@ public class ESAdapter implements OuterAdapter { for (ESSyncConfig configTmp : esSyncConfig.values()) { // 取所有的destination为task的配置 if (configTmp.getDestination().equals(task)) { - ESEtlService esEtlService = new ESEtlService(transportClient, configTmp); + ESEtlService esEtlService = new ESEtlService(esConnection, configTmp); EtlResult etlRes = esEtlService.importData(params); if (!etlRes.getSucceeded()) { resSuccess = false; @@ -220,10 +219,8 @@ public class ESAdapter implements OuterAdapter { public Map count(String task) { ESSyncConfig config = esSyncConfig.get(task); ESMapping mapping = config.getEsMapping(); - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setSize(0) - .get(); + SearchResponse response = this.esConnection.new ESSearchRequest(mapping.get_index(), mapping.get_type()).size(0) + .getResponse(); long rowCount = response.getHits().getTotalHits(); Map res = new LinkedHashMap<>(); @@ -237,8 +234,8 @@ public class ESAdapter implements OuterAdapter { if (esConfigMonitor != null) { esConfigMonitor.destroy(); } - if (transportClient != null) { - transportClient.close(); + if (esConnection != null) { + esConnection.close(); } } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java index f481a5bf..218a32e8 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESEtlService.java @@ -11,12 +11,8 @@ import javax.sql.DataSource; import org.apache.commons.lang.StringUtils; import org.elasticsearch.action.bulk.BulkItemResponse; -import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; -import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; import org.elasticsearch.search.SearchHit; @@ -24,6 +20,11 @@ import org.elasticsearch.search.SearchHit; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; +import com.alibaba.otter.canal.client.adapter.es.support.ESConnection; +import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESBulkRequest; +import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESIndexRequest; +import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESSearchRequest; +import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.ESUpdateRequest; import com.alibaba.otter.canal.client.adapter.es.support.ESTemplate; import com.alibaba.otter.canal.client.adapter.support.AbstractEtlService; import com.alibaba.otter.canal.client.adapter.support.AdapterConfig; @@ -38,14 +39,14 @@ import com.alibaba.otter.canal.client.adapter.support.Util; */ public class ESEtlService extends AbstractEtlService { - private TransportClient transportClient; - private ESTemplate esTemplate; - private ESSyncConfig config; + private ESConnection esConnection; + private ESTemplate esTemplate; + private ESSyncConfig config; - public ESEtlService(TransportClient transportClient, ESSyncConfig config){ + public ESEtlService(ESConnection esConnection, ESSyncConfig config){ super("ES", config); - this.transportClient = transportClient; - this.esTemplate = new ESTemplate(transportClient); + this.esConnection = esConnection; + this.esTemplate = new ESTemplate(esConnection); this.config = config; } @@ -79,7 +80,7 @@ public class ESEtlService extends AbstractEtlService { Util.sqlRS(ds, sql, values, rs -> { int count = 0; try { - BulkRequestBuilder bulkRequestBuilder = transportClient.prepareBulk(); + ESBulkRequest esBulkRequest = this.esConnection.new ESBulkRequest(); long batchBegin = System.currentTimeMillis(); while (rs.next()) { @@ -132,41 +133,41 @@ public class ESEtlService extends AbstractEtlService { if (idVal != null) { String parentVal = (String) esFieldData.remove("$parent_routing"); if (mapping.isUpsert()) { - UpdateRequestBuilder updateRequestBuilder = transportClient - .prepareUpdate(mapping.get_index(), mapping.get_type(), idVal.toString()) - .setDoc(esFieldData) - .setDocAsUpsert(true); + ESUpdateRequest esUpdateRequest = this.esConnection.new ESUpdateRequest( + mapping.get_index(), + mapping.get_type(), + idVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); + if (StringUtils.isNotEmpty(parentVal)) { - updateRequestBuilder.setRouting(parentVal); + esUpdateRequest.setRouting(parentVal); } - bulkRequestBuilder.add(updateRequestBuilder); + + esBulkRequest.add(esUpdateRequest); } else { - IndexRequestBuilder indexRequestBuilder = transportClient - .prepareIndex(mapping.get_index(), mapping.get_type(), idVal.toString()) - .setSource(esFieldData); + ESIndexRequest esIndexRequest = this.esConnection.new ESIndexRequest(mapping + .get_index(), mapping.get_type(), idVal.toString()).setSource(esFieldData); if (StringUtils.isNotEmpty(parentVal)) { - indexRequestBuilder.setRouting(parentVal); + esIndexRequest.setRouting(parentVal); } - bulkRequestBuilder.add(indexRequestBuilder); + esBulkRequest.add(esIndexRequest); } } else { idVal = esFieldData.get(mapping.getPk()); - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) - .setSize(10000) - .get(); + ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), + mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), idVal)) + .size(10000); + SearchResponse response = esSearchRequest.getResponse(); for (SearchHit hit : response.getHits()) { - bulkRequestBuilder.add( - transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) - .setDoc(esFieldData)); + ESUpdateRequest esUpdateRequest = this.esConnection.new ESUpdateRequest(mapping + .get_index(), mapping.get_type(), hit.getId()).setDoc(esFieldData); + esBulkRequest.add(esUpdateRequest); } } - if (bulkRequestBuilder.numberOfActions() % mapping.getCommitBatch() == 0 - && bulkRequestBuilder.numberOfActions() > 0) { + if (esBulkRequest.numberOfActions() % mapping.getCommitBatch() == 0 + && esBulkRequest.numberOfActions() > 0) { long esBatchBegin = System.currentTimeMillis(); - BulkResponse rp = bulkRequestBuilder.execute().actionGet(); + BulkResponse rp = esBulkRequest.bulk(); if (rp.hasFailures()) { this.processFailBulkResponse(rp); } @@ -175,19 +176,19 @@ public class ESEtlService extends AbstractEtlService { logger.trace("全量数据批量导入批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", (System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - esBatchBegin), - bulkRequestBuilder.numberOfActions(), + esBulkRequest.numberOfActions(), mapping.get_index()); } batchBegin = System.currentTimeMillis(); - bulkRequestBuilder = transportClient.prepareBulk(); + esBulkRequest.resetBulk(); } count++; impCount.incrementAndGet(); } - if (bulkRequestBuilder.numberOfActions() > 0) { + if (esBulkRequest.numberOfActions() > 0) { long esBatchBegin = System.currentTimeMillis(); - BulkResponse rp = bulkRequestBuilder.execute().actionGet(); + BulkResponse rp = esBulkRequest.bulk(); if (rp.hasFailures()) { this.processFailBulkResponse(rp); } @@ -195,7 +196,7 @@ public class ESEtlService extends AbstractEtlService { logger.trace("全量数据批量导入最后批次耗时: {}, es执行时间: {}, 批次大小: {}, index; {}", (System.currentTimeMillis() - batchBegin), (System.currentTimeMillis() - esBatchBegin), - bulkRequestBuilder.numberOfActions(), + esBulkRequest.numberOfActions(), mapping.get_index()); } } diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESConnection.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESConnection.java new file mode 100644 index 00000000..634f94c4 --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESConnection.java @@ -0,0 +1,468 @@ +package com.alibaba.otter.canal.client.adapter.es.support; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.Map; + +import org.apache.commons.lang.StringUtils; +import org.apache.http.HttpHost; +import org.apache.http.auth.AuthScope; +import org.apache.http.auth.UsernamePasswordCredentials; +import org.apache.http.client.CredentialsProvider; +import org.apache.http.impl.client.BasicCredentialsProvider; +import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; +import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; +import org.elasticsearch.action.bulk.BulkRequest; +import org.elasticsearch.action.bulk.BulkRequestBuilder; +import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.action.delete.DeleteRequest; +import org.elasticsearch.action.delete.DeleteRequestBuilder; +import org.elasticsearch.action.index.IndexRequest; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.search.SearchRequest; +import org.elasticsearch.action.search.SearchRequestBuilder; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.action.update.UpdateRequest; +import org.elasticsearch.action.update.UpdateRequestBuilder; +import org.elasticsearch.client.*; +import org.elasticsearch.client.transport.TransportClient; +import org.elasticsearch.cluster.metadata.MappingMetaData; +import org.elasticsearch.common.collect.ImmutableOpenMap; +import org.elasticsearch.common.settings.Settings; +import org.elasticsearch.common.transport.TransportAddress; +import org.elasticsearch.index.query.QueryBuilder; +import org.elasticsearch.search.builder.SearchSourceBuilder; +import org.elasticsearch.transport.client.PreBuiltTransportClient; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * ES 连接器, Transport Rest 两种方式 + * + * @author rewerma 2019-08-01 + * @version 1.0.0 + */ +public class ESConnection { + + private static final Logger logger = LoggerFactory.getLogger(ESConnection.class); + + public enum ESClientMode { + TRANSPORT, REST + } + + private ESClientMode mode; + + private TransportClient transportClient; + + private RestHighLevelClient restHighLevelClient; + + public ESConnection(String[] hosts, Map properties, ESClientMode mode) throws UnknownHostException{ + this.mode = mode; + if (mode == ESClientMode.TRANSPORT) { + Settings.Builder settingBuilder = Settings.builder(); + properties.forEach(settingBuilder::put); + Settings settings = settingBuilder.build(); + transportClient = new PreBuiltTransportClient(settings); + for (String host : hosts) { + int i = host.indexOf(":"); + transportClient.addTransportAddress(new TransportAddress(InetAddress.getByName(host.substring(0, i)), + Integer.parseInt(host.substring(i + 1)))); + } + } else { + HttpHost[] httpHosts = new HttpHost[hosts.length]; + for (int i = 0; i < hosts.length; i++) { + String host = hosts[i]; + int j = host.indexOf(":"); + HttpHost httpHost = new HttpHost(InetAddress.getByName(host.substring(0, j)), + Integer.parseInt(host.substring(j + 1))); + httpHosts[i] = httpHost; + } + RestClientBuilder restClientBuilder = RestClient.builder(httpHosts); + String nameAndPwd = properties.get("security.auth"); + if (StringUtils.isNotEmpty(nameAndPwd) && nameAndPwd.contains(":")) { + String[] nameAndPwdArr = nameAndPwd.split(":"); + final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); + credentialsProvider.setCredentials(AuthScope.ANY, + new UsernamePasswordCredentials(nameAndPwdArr[0], nameAndPwdArr[1])); + restClientBuilder.setHttpClientConfigCallback( + httpClientBuilder -> httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider)); + } + restHighLevelClient = new RestHighLevelClient(restClientBuilder); + } + } + + public void close() { + if (mode == ESClientMode.TRANSPORT) { + transportClient.close(); + } else { + try { + restHighLevelClient.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + public MappingMetaData getMapping(String index, String type) { + MappingMetaData mappingMetaData = null; + if (mode == ESClientMode.TRANSPORT) { + ImmutableOpenMap mappings; + try { + mappings = transportClient.admin() + .cluster() + .prepareState() + .execute() + .actionGet() + .getState() + .getMetaData() + .getIndices() + .get(index) + .getMappings(); + } catch (NullPointerException e) { + throw new IllegalArgumentException("Not found the mapping info of index: " + index); + } + mappingMetaData = mappings.get(type); + + } else { + ImmutableOpenMap> mappings; + try { + GetMappingsRequest request = new GetMappingsRequest(); + request.indices(index); + GetMappingsResponse response; + // try { + // response = restHighLevelClient + // .indices() + // .getMapping(request, RequestOptions.DEFAULT); + // // 6.4以下版本直接使用该接口会报错 + // } catch (Exception e) { + // logger.warn("Low ElasticSearch version for getMapping"); + response = RestHighLevelClientExt.getMapping(restHighLevelClient, request, RequestOptions.DEFAULT); + // } + + mappings = response.mappings(); + } catch (NullPointerException e) { + throw new IllegalArgumentException("Not found the mapping info of index: " + index); + } catch (IOException e) { + logger.error(e.getMessage(), e); + return null; + } + mappingMetaData = mappings.get(index).get(type); + } + return mappingMetaData; + } + + public class ESIndexRequest { + + private IndexRequestBuilder indexRequestBuilder; + + private IndexRequest indexRequest; + + public ESIndexRequest(String index, String type, String id){ + if (mode == ESClientMode.TRANSPORT) { + indexRequestBuilder = transportClient.prepareIndex(index, type, id); + } else { + indexRequest = new IndexRequest(index, type, id); + } + } + + public ESIndexRequest setSource(Map source) { + if (mode == ESClientMode.TRANSPORT) { + indexRequestBuilder.setSource(source); + } else { + indexRequest.source(source); + } + return this; + } + + public ESIndexRequest setRouting(String routing) { + if (mode == ESClientMode.TRANSPORT) { + indexRequestBuilder.setRouting(routing); + } else { + indexRequest.routing(routing); + } + return this; + } + + public IndexRequestBuilder getIndexRequestBuilder() { + return indexRequestBuilder; + } + + public void setIndexRequestBuilder(IndexRequestBuilder indexRequestBuilder) { + this.indexRequestBuilder = indexRequestBuilder; + } + + public IndexRequest getIndexRequest() { + return indexRequest; + } + + public void setIndexRequest(IndexRequest indexRequest) { + this.indexRequest = indexRequest; + } + } + + public class ESUpdateRequest { + + private UpdateRequestBuilder updateRequestBuilder; + + private UpdateRequest updateRequest; + + public ESUpdateRequest(String index, String type, String id){ + if (mode == ESClientMode.TRANSPORT) { + updateRequestBuilder = transportClient.prepareUpdate(index, type, id); + } else { + updateRequest = new UpdateRequest(index, type, id); + } + } + + public ESUpdateRequest setDoc(Map source) { + if (mode == ESClientMode.TRANSPORT) { + updateRequestBuilder.setDoc(source); + } else { + updateRequest.doc(source); + } + return this; + } + + public ESUpdateRequest setDocAsUpsert(boolean shouldUpsertDoc) { + if (mode == ESClientMode.TRANSPORT) { + updateRequestBuilder.setDocAsUpsert(shouldUpsertDoc); + } else { + updateRequest.docAsUpsert(shouldUpsertDoc); + } + return this; + } + + public ESUpdateRequest setRouting(String routing) { + if (mode == ESClientMode.TRANSPORT) { + updateRequestBuilder.setRouting(routing); + } else { + updateRequest.routing(routing); + } + return this; + } + + public UpdateRequestBuilder getUpdateRequestBuilder() { + return updateRequestBuilder; + } + + public void setUpdateRequestBuilder(UpdateRequestBuilder updateRequestBuilder) { + this.updateRequestBuilder = updateRequestBuilder; + } + + public UpdateRequest getUpdateRequest() { + return updateRequest; + } + + public void setUpdateRequest(UpdateRequest updateRequest) { + this.updateRequest = updateRequest; + } + } + + public class ESDeleteRequest { + + private DeleteRequestBuilder deleteRequestBuilder; + + private DeleteRequest deleteRequest; + + public ESDeleteRequest(String index, String type, String id){ + if (mode == ESClientMode.TRANSPORT) { + deleteRequestBuilder = transportClient.prepareDelete(index, type, id); + } else { + deleteRequest = new DeleteRequest(index, type, id); + } + } + + public DeleteRequestBuilder getDeleteRequestBuilder() { + return deleteRequestBuilder; + } + + public void setDeleteRequestBuilder(DeleteRequestBuilder deleteRequestBuilder) { + this.deleteRequestBuilder = deleteRequestBuilder; + } + + public DeleteRequest getDeleteRequest() { + return deleteRequest; + } + + public void setDeleteRequest(DeleteRequest deleteRequest) { + this.deleteRequest = deleteRequest; + } + } + + public class ESSearchRequest { + + private SearchRequestBuilder searchRequestBuilder; + + private SearchRequest searchRequest; + + private SearchSourceBuilder sourceBuilder; + + public ESSearchRequest(String index, String... types){ + if (mode == ESClientMode.TRANSPORT) { + searchRequestBuilder = transportClient.prepareSearch(index).setTypes(types); + } else { + searchRequest = new SearchRequest(index).types(types); + sourceBuilder = new SearchSourceBuilder(); + } + } + + public ESSearchRequest setQuery(QueryBuilder queryBuilder) { + if (mode == ESClientMode.TRANSPORT) { + searchRequestBuilder.setQuery(queryBuilder); + } else { + sourceBuilder.query(queryBuilder); + } + return this; + } + + public ESSearchRequest size(int size) { + if (mode == ESClientMode.TRANSPORT) { + searchRequestBuilder.setSize(size); + } else { + sourceBuilder.size(size); + } + return this; + } + + public SearchResponse getResponse() { + if (mode == ESClientMode.TRANSPORT) { + return searchRequestBuilder.get(); + } else { + searchRequest.source(sourceBuilder); + try { + return restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + public SearchRequestBuilder getSearchRequestBuilder() { + return searchRequestBuilder; + } + + public void setSearchRequestBuilder(SearchRequestBuilder searchRequestBuilder) { + this.searchRequestBuilder = searchRequestBuilder; + } + + public SearchRequest getSearchRequest() { + return searchRequest; + } + + public void setSearchRequest(SearchRequest searchRequest) { + this.searchRequest = searchRequest; + } + } + + public class ESBulkRequest { + + private BulkRequestBuilder bulkRequestBuilder; + + private BulkRequest bulkRequest; + + public ESBulkRequest(){ + if (mode == ESClientMode.TRANSPORT) { + bulkRequestBuilder = transportClient.prepareBulk(); + } else { + bulkRequest = new BulkRequest(); + } + } + + public void resetBulk() { + if (mode == ESClientMode.TRANSPORT) { + bulkRequestBuilder = transportClient.prepareBulk(); + } else { + bulkRequest = new BulkRequest(); + } + } + + public ESBulkRequest add(ESIndexRequest esIndexRequest) { + if (mode == ESClientMode.TRANSPORT) { + bulkRequestBuilder.add(esIndexRequest.indexRequestBuilder); + } else { + bulkRequest.add(esIndexRequest.indexRequest); + } + return this; + } + + public ESBulkRequest add(ESUpdateRequest esUpdateRequest) { + if (mode == ESClientMode.TRANSPORT) { + bulkRequestBuilder.add(esUpdateRequest.updateRequestBuilder); + } else { + bulkRequest.add(esUpdateRequest.updateRequest); + } + return this; + } + + public ESBulkRequest add(ESDeleteRequest esDeleteRequest) { + if (mode == ESClientMode.TRANSPORT) { + bulkRequestBuilder.add(esDeleteRequest.deleteRequestBuilder); + } else { + bulkRequest.add(esDeleteRequest.deleteRequest); + } + return this; + } + + public int numberOfActions() { + if (mode == ESClientMode.TRANSPORT) { + return bulkRequestBuilder.numberOfActions(); + } else { + return bulkRequest.numberOfActions(); + } + } + + public BulkResponse bulk() { + if (mode == ESClientMode.TRANSPORT) { + return bulkRequestBuilder.execute().actionGet(); + } else { + try { + return restHighLevelClient.bulk(bulkRequest); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + } + + public BulkRequestBuilder getBulkRequestBuilder() { + return bulkRequestBuilder; + } + + public void setBulkRequestBuilder(BulkRequestBuilder bulkRequestBuilder) { + this.bulkRequestBuilder = bulkRequestBuilder; + } + + public BulkRequest getBulkRequest() { + return bulkRequest; + } + + public void setBulkRequest(BulkRequest bulkRequest) { + this.bulkRequest = bulkRequest; + } + } + + // ------ get/set ------ + public ESClientMode getMode() { + return mode; + } + + public void setMode(ESClientMode mode) { + this.mode = mode; + } + + public TransportClient getTransportClient() { + return transportClient; + } + + public void setTransportClient(TransportClient transportClient) { + this.transportClient = transportClient; + } + + public RestHighLevelClient getRestHighLevelClient() { + return restHighLevelClient; + } + + public void setRestHighLevelClient(RestHighLevelClient restHighLevelClient) { + this.restHighLevelClient = restHighLevelClient; + } +} diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java index 1aaa14c9..68187d62 100644 --- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java +++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java @@ -10,14 +10,9 @@ import javax.sql.DataSource; import org.apache.commons.lang.StringUtils; import org.elasticsearch.action.bulk.BulkItemResponse; -import org.elasticsearch.action.bulk.BulkRequestBuilder; import org.elasticsearch.action.bulk.BulkResponse; -import org.elasticsearch.action.index.IndexRequestBuilder; import org.elasticsearch.action.search.SearchResponse; -import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.elasticsearch.client.transport.TransportClient; import org.elasticsearch.cluster.metadata.MappingMetaData; -import org.elasticsearch.common.collect.ImmutableOpenMap; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.rest.RestStatus; @@ -30,6 +25,7 @@ import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig.ESMapping; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.ColumnItem; import com.alibaba.otter.canal.client.adapter.es.config.SchemaItem.FieldItem; +import com.alibaba.otter.canal.client.adapter.es.support.ESConnection.*; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Util; @@ -41,65 +37,65 @@ import com.alibaba.otter.canal.client.adapter.support.Util; */ public class ESTemplate { - private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class); + private static final Logger logger = LoggerFactory.getLogger(ESTemplate.class); - private static final int MAX_BATCH_SIZE = 1000; + private static final int MAX_BATCH_SIZE = 1000; - private TransportClient transportClient; + private ESConnection esConnection; - private BulkRequestBuilder bulkRequestBuilder; + private ESBulkRequest esBulkRequest; - public ESTemplate(TransportClient transportClient){ - this.transportClient = transportClient; - this.bulkRequestBuilder = transportClient.prepareBulk(); + public ESTemplate(ESConnection esConnection) { + this.esConnection = esConnection; + this.esBulkRequest = this.esConnection.new ESBulkRequest(); } - public BulkRequestBuilder getBulk() { - return bulkRequestBuilder; + public ESBulkRequest getBulk() { + return esBulkRequest; } public void resetBulkRequestBuilder() { - this.bulkRequestBuilder = this.transportClient.prepareBulk(); + this.esBulkRequest.resetBulk(); } /** * 插入数据 * - * @param mapping 配置对象 - * @param pkVal 主键值 + * @param mapping 配置对象 + * @param pkVal 主键值 * @param esFieldData 数据Map */ public void insert(ESMapping mapping, Object pkVal, Map esFieldData) { if (mapping.get_id() != null) { String parentVal = (String) esFieldData.remove("$parent_routing"); if (mapping.isUpsert()) { - UpdateRequestBuilder updateRequestBuilder = transportClient - .prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData) - .setDocAsUpsert(true); + ESUpdateRequest updateRequest = esConnection.new ESUpdateRequest(mapping.get_index(), + mapping.get_type(), + pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); if (StringUtils.isNotEmpty(parentVal)) { - updateRequestBuilder.setRouting(parentVal); + updateRequest.setRouting(parentVal); } - getBulk().add(updateRequestBuilder); + getBulk().add(updateRequest); } else { - IndexRequestBuilder indexRequestBuilder = transportClient - .prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setSource(esFieldData); + ESIndexRequest indexRequest = esConnection.new ESIndexRequest(mapping.get_index(), + mapping.get_type(), + pkVal.toString()).setSource(esFieldData); if (StringUtils.isNotEmpty(parentVal)) { - indexRequestBuilder.setRouting(parentVal); + indexRequest.setRouting(parentVal); } - getBulk().add(indexRequestBuilder); + getBulk().add(indexRequest); } commitBulk(); } else { - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .setSize(10000) - .get(); + ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), + mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000); + SearchResponse response = esSearchRequest.getResponse(); + for (SearchHit hit : response.getHits()) { - getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) - .setDoc(esFieldData)); + ESUpdateRequest esUpdateRequest = this.esConnection.new ESUpdateRequest(mapping.get_index(), + mapping.get_type(), + hit.getId()).setDoc(esFieldData); + getBulk().add(esUpdateRequest); commitBulk(); } } @@ -109,8 +105,8 @@ public class ESTemplate { /** * 根据主键更新数据 * - * @param mapping 配置对象 - * @param pkVal 主键值 + * @param mapping 配置对象 + * @param pkVal 主键值 * @param esFieldData 数据Map */ public void update(ESMapping mapping, Object pkVal, Map esFieldData) { @@ -123,8 +119,8 @@ public class ESTemplate { /** * update by query * - * @param config 配置对象 - * @param paramsTmp sql查询条件 + * @param config 配置对象 + * @param paramsTmp sql查询条件 * @param esFieldData 数据Map */ public void updateByQuery(ESSyncConfig config, Map paramsTmp, Map esFieldData) { @@ -143,7 +139,7 @@ public class ESTemplate { sql.append("_v.").append(fieldName).append("=? AND "); values.add(value); }); - //TODO 直接外部包裹sql会导致全表扫描性能低, 待优化拼接内部where条件 + // TODO 直接外部包裹sql会导致全表扫描性能低, 待优化拼接内部where条件 int len = sql.length(); sql.delete(len - 4, len); Integer syncCount = (Integer) Util.sqlRS(ds, sql.toString(), values, rs -> { @@ -168,23 +164,26 @@ public class ESTemplate { /** * 通过主键删除数据 * - * @param mapping 配置对象 - * @param pkVal 主键值 + * @param mapping 配置对象 + * @param pkVal 主键值 * @param esFieldData 数据Map */ public void delete(ESMapping mapping, Object pkVal, Map esFieldData) { if (mapping.get_id() != null) { - getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString())); + ESDeleteRequest esDeleteRequest = this.esConnection.new ESDeleteRequest(mapping.get_index(), + mapping.get_type(), + pkVal.toString()); + getBulk().add(esDeleteRequest); commitBulk(); } else { - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .setSize(10000) - .get(); + ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), + mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000); + SearchResponse response = esSearchRequest.getResponse(); for (SearchHit hit : response.getHits()) { - getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) - .setDoc(esFieldData)); + ESUpdateRequest esUpdateRequest = this.esConnection.new ESUpdateRequest(mapping.get_index(), + mapping.get_type(), + hit.getId()).setDoc(esFieldData); + getBulk().add(esUpdateRequest); commitBulk(); } } @@ -196,7 +195,7 @@ public class ESTemplate { */ public void commit() { if (getBulk().numberOfActions() > 0) { - BulkResponse response = getBulk().execute().actionGet(); + BulkResponse response = getBulk().bulk(); if (response.hasFailures()) { for (BulkItemResponse itemResponse : response.getItems()) { if (!itemResponse.isFailed()) { @@ -227,32 +226,31 @@ public class ESTemplate { if (mapping.get_id() != null) { String parentVal = (String) esFieldData.remove("$parent_routing"); if (mapping.isUpsert()) { - UpdateRequestBuilder updateRequestBuilder = transportClient - .prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData) - .setDocAsUpsert(true); + ESUpdateRequest esUpdateRequest = this.esConnection.new ESUpdateRequest(mapping.get_index(), + mapping.get_type(), + pkVal.toString()).setDoc(esFieldData).setDocAsUpsert(true); if (StringUtils.isNotEmpty(parentVal)) { - updateRequestBuilder.setRouting(parentVal); + esUpdateRequest.setRouting(parentVal); } - getBulk().add(updateRequestBuilder); + getBulk().add(esUpdateRequest); } else { - UpdateRequestBuilder updateRequestBuilder = transportClient - .prepareUpdate(mapping.get_index(), mapping.get_type(), pkVal.toString()) - .setDoc(esFieldData); + ESUpdateRequest esUpdateRequest = this.esConnection.new ESUpdateRequest(mapping.get_index(), + mapping.get_type(), + pkVal.toString()).setDoc(esFieldData); if (StringUtils.isNotEmpty(parentVal)) { - updateRequestBuilder.setRouting(parentVal); + esUpdateRequest.setRouting(parentVal); } - getBulk().add(updateRequestBuilder); + getBulk().add(esUpdateRequest); } } else { - SearchResponse response = transportClient.prepareSearch(mapping.get_index()) - .setTypes(mapping.get_type()) - .setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)) - .setSize(10000) - .get(); + ESSearchRequest esSearchRequest = this.esConnection.new ESSearchRequest(mapping.get_index(), + mapping.get_type()).setQuery(QueryBuilders.termQuery(mapping.getPk(), pkVal)).size(10000); + SearchResponse response = esSearchRequest.getResponse(); for (SearchHit hit : response.getHits()) { - getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId()) - .setDoc(esFieldData)); + ESUpdateRequest esUpdateRequest = this.esConnection.new ESUpdateRequest(mapping.get_index(), + mapping.get_type(), + hit.getId()).setDoc(esFieldData); + getBulk().add(esUpdateRequest); } } } @@ -291,7 +289,7 @@ public class ESTemplate { } if (!fieldItem.getFieldName().equals(mapping.get_id()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { + && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } } @@ -329,9 +327,9 @@ public class ESTemplate { for (ColumnItem columnItem : fieldItem.getColumnItems()) { if (dmlOld.containsKey(columnItem.getColumnName()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { + && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), - getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); + getValFromRS(mapping, resultSet, fieldItem.getFieldName(), fieldItem.getFieldName())); break; } } @@ -363,8 +361,8 @@ public class ESTemplate { /** * 将dml的data转换为es的data * - * @param mapping 配置mapping - * @param dmlData dml data + * @param mapping 配置mapping + * @param dmlData dml data * @param esFieldData es data * @return 返回 id 值 */ @@ -382,7 +380,7 @@ public class ESTemplate { } if (!fieldItem.getFieldName().equals(mapping.get_id()) - && !mapping.getSkips().contains(fieldItem.getFieldName())) { + && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), value); } } @@ -395,8 +393,8 @@ public class ESTemplate { /** * 将dml的data, old转换为es的data * - * @param mapping 配置mapping - * @param dmlData dml data + * @param mapping 配置mapping + * @param dmlData dml data * @param esFieldData es data * @return 返回 id 值 */ @@ -414,7 +412,7 @@ public class ESTemplate { if (dmlOld.containsKey(columnName) && !mapping.getSkips().contains(fieldItem.getFieldName())) { esFieldData.put(Util.cleanColumn(fieldItem.getFieldName()), - getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); + getValFromData(mapping, dmlData, fieldItem.getFieldName(), columnName)); } } @@ -435,9 +433,9 @@ public class ESTemplate { Object parentVal; try { parentVal = getValFromRS(mapping, - resultSet, - parentFieldItem.getFieldName(), - parentFieldItem.getFieldName()); + resultSet, + parentFieldItem.getFieldName(), + parentFieldItem.getFieldName()); } catch (SQLException e) { throw new RuntimeException(e); } @@ -482,7 +480,7 @@ public class ESTemplate { /** * 获取es mapping中的属性类型 * - * @param mapping mapping配置 + * @param mapping mapping配置 * @param fieldName 属性名 * @return 类型 */ @@ -490,23 +488,11 @@ public class ESTemplate { private String getEsType(ESMapping mapping, String fieldName) { String key = mapping.get_index() + "-" + mapping.get_type(); Map fieldType = esFieldTypes.get(key); - if (fieldType == null) { - ImmutableOpenMap mappings; - try { - mappings = transportClient.admin() - .cluster() - .prepareState() - .execute() - .actionGet() - .getState() - .getMetaData() - .getIndices() - .get(mapping.get_index()) - .getMappings(); - } catch (NullPointerException e) { - throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); - } - MappingMetaData mappingMetaData = mappings.get(mapping.get_type()); + if (fieldType != null) { + return fieldType.get(fieldName); + } else { + MappingMetaData mappingMetaData = esConnection.getMapping(mapping.get_index(), mapping.get_type()); + if (mappingMetaData == null) { throw new IllegalArgumentException("Not found the mapping info of index: " + mapping.get_index()); } @@ -524,8 +510,9 @@ public class ESTemplate { } } esFieldTypes.put(key, fieldType); - } - return fieldType.get(fieldName); + return fieldType.get(fieldName); + } } + } diff --git a/client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RequestConvertersExt.java b/client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RequestConvertersExt.java new file mode 100644 index 00000000..7ac2d5ce --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RequestConvertersExt.java @@ -0,0 +1,30 @@ +package org.elasticsearch.client; + +import org.apache.http.client.methods.HttpGet; +import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; +import org.elasticsearch.common.Strings; + +import java.io.IOException; + +/** + * RequestConverters扩展 + * + * @author rewerma 2019-08-01 + * @version 1.0.0 + */ +public class RequestConvertersExt { + + /** + * 修改 getMappings 去掉request参数 + * + * @param getMappingsRequest + * @return + * @throws IOException + */ + static Request getMappings(GetMappingsRequest getMappingsRequest) throws IOException { + String[] indices = getMappingsRequest.indices() == null ? Strings.EMPTY_ARRAY : getMappingsRequest.indices(); + String[] types = getMappingsRequest.types() == null ? Strings.EMPTY_ARRAY : getMappingsRequest.types(); + + return new Request(HttpGet.METHOD_NAME, RequestConverters.endpoint(indices, "_mapping", types)); + } +} diff --git a/client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RestHighLevelClientExt.java b/client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RestHighLevelClientExt.java new file mode 100644 index 00000000..a55295ef --- /dev/null +++ b/client-adapter/elasticsearch/src/main/java/org/elasticsearch/client/RestHighLevelClientExt.java @@ -0,0 +1,29 @@ +package org.elasticsearch.client; + +import static java.util.Collections.emptySet; + +import java.io.IOException; + +import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; +import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; + +/** + * RestHighLevelClient扩展 + * + * @author rewerma 2019-08-01 + * @version 1.0.0 + */ +public class RestHighLevelClientExt { + + public static GetMappingsResponse getMapping(RestHighLevelClient restHighLevelClient, + GetMappingsRequest getMappingsRequest, + RequestOptions options) throws IOException { + return restHighLevelClient.performRequestAndParseEntity(getMappingsRequest, + RequestConvertersExt::getMappings, + options, + GetMappingsResponse::fromXContent, + emptySet()); + + } + +} diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java index 5c35cc7d..43332fcf 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSub2Test.java @@ -18,6 +18,7 @@ import com.alibaba.otter.canal.client.adapter.es.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; + @Ignore public class LabelSyncJoinSub2Test { @@ -59,7 +60,10 @@ public class LabelSyncJoinSub2Test { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("b;a_", response.getSource().get("_labels")); } @@ -97,7 +101,10 @@ public class LabelSyncJoinSub2Test { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("b;aa_", response.getSource().get("_labels")); } @@ -129,7 +136,10 @@ public class LabelSyncJoinSub2Test { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("b_", response.getSource().get("_labels")); } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java index 3dd7d262..a2387c76 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/LabelSyncJoinSubTest.java @@ -18,6 +18,7 @@ import com.alibaba.otter.canal.client.adapter.es.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; + @Ignore public class LabelSyncJoinSubTest { @@ -59,7 +60,10 @@ public class LabelSyncJoinSubTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("b;a", response.getSource().get("_labels")); } @@ -97,7 +101,10 @@ public class LabelSyncJoinSubTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("b;aa", response.getSource().get("_labels")); } @@ -129,7 +136,10 @@ public class LabelSyncJoinSubTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("b", response.getSource().get("_labels")); } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java index e998115e..85bc2d00 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOne2Test.java @@ -18,6 +18,7 @@ import com.alibaba.otter.canal.client.adapter.es.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; + @Ignore public class RoleSyncJoinOne2Test { @@ -57,7 +58,10 @@ public class RoleSyncJoinOne2Test { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("admin_", response.getSource().get("_role_name")); } @@ -94,7 +98,10 @@ public class RoleSyncJoinOne2Test { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("admin3_", response.getSource().get("_role_name")); } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java index 7813e808..52c6a887 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/RoleSyncJoinOneTest.java @@ -58,7 +58,7 @@ public class RoleSyncJoinOneTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection().getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin", response.getSource().get("_role_name")); } @@ -95,7 +95,7 @@ public class RoleSyncJoinOneTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection().getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin2", response.getSource().get("_role_name")); } @@ -133,7 +133,7 @@ public class RoleSyncJoinOneTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection().getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("operator", response.getSource().get("_role_name")); Common.sqlExe(ds, "update user set role_id=1 where id=1"); @@ -158,7 +158,7 @@ public class RoleSyncJoinOneTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml2); - GetResponse response2 = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response2 = esAdapter.getEsConnection().getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertEquals("admin2", response2.getSource().get("_role_name")); } @@ -190,7 +190,8 @@ public class RoleSyncJoinOneTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); Assert.assertNull(response.getSource().get("_role_name")); } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java index 55fd6cf7..18677afc 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncJoinOneTest.java @@ -18,6 +18,7 @@ import com.alibaba.otter.canal.client.adapter.es.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; + @Ignore public class UserSyncJoinOneTest { @@ -59,7 +60,10 @@ public class UserSyncJoinOneTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("Eric_", response.getSource().get("_name")); } @@ -95,7 +99,10 @@ public class UserSyncJoinOneTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("Eric2_", response.getSource().get("_name")); } } diff --git a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java index 7b359fa9..600db295 100644 --- a/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java +++ b/client-adapter/elasticsearch/src/test/java/com/alibaba/otter/canal/client/adapter/es/test/sync/UserSyncSingleTest.java @@ -15,6 +15,7 @@ import org.junit.Test; import com.alibaba.otter.canal.client.adapter.es.ESAdapter; import com.alibaba.otter.canal.client.adapter.es.config.ESSyncConfig; import com.alibaba.otter.canal.client.adapter.support.Dml; + @Ignore public class UserSyncSingleTest { @@ -52,7 +53,10 @@ public class UserSyncSingleTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("Eric", response.getSource().get("_name")); } @@ -85,7 +89,10 @@ public class UserSyncSingleTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertEquals("Eric2", response.getSource().get("_name")); } @@ -115,7 +122,10 @@ public class UserSyncSingleTest { esAdapter.getEsSyncService().sync(esSyncConfigs.values(), dml); - GetResponse response = esAdapter.getTransportClient().prepareGet("mytest_user", "_doc", "1").get(); + GetResponse response = esAdapter.getEsConnection() + .getTransportClient() + .prepareGet("mytest_user", "_doc", "1") + .get(); Assert.assertNull(response.getSource()); } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java index 0f1f821d..34beb094 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/monitor/remote/DbRemoteConfigLoader.java @@ -1,251 +1,251 @@ -package com.alibaba.otter.canal.adapter.launcher.monitor.remote; - -import java.io.FileOutputStream; -import java.io.OutputStreamWriter; -import java.nio.charset.StandardCharsets; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.SQLException; -import java.sql.Statement; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; - -import org.apache.commons.lang.StringUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.alibaba.druid.pool.DruidDataSource; -import com.alibaba.otter.canal.common.utils.CommonUtils; -import com.alibaba.otter.canal.common.utils.NamedThreadFactory; -import com.google.common.base.Joiner; -import com.google.common.collect.MapMaker; - -/** - * 基于数据库的远程配置装载器 - * - * @author rewerma 2019-01-25 下午05:20:16 - * @version 1.0.0 - */ -public class DbRemoteConfigLoader implements RemoteConfigLoader { - - private static final Logger logger = LoggerFactory.getLogger(DbRemoteConfigLoader.class); - - private DruidDataSource dataSource; - - private volatile long currentConfigTimestamp = 0; - private Map remoteAdapterConfigs = new MapMaker().makeMap(); - - private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, - new NamedThreadFactory("remote-adapter-config-scan")); - - private RemoteAdapterMonitor remoteAdapterMonitor = new RemoteAdapterMonitorImpl(); - - public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){ - dataSource = new DruidDataSource(); - if (StringUtils.isEmpty(driverName)) { - driverName = "com.mysql.jdbc.Driver"; - } - dataSource.setDriverClassName(driverName); - dataSource.setUrl(jdbcUrl); - dataSource.setUsername(jdbcUsername); - dataSource.setPassword(jdbcPassword); - dataSource.setInitialSize(1); - dataSource.setMinIdle(1); - dataSource.setMaxActive(1); - dataSource.setMaxWait(60000); - dataSource.setTimeBetweenEvictionRunsMillis(60000); - dataSource.setMinEvictableIdleTimeMillis(300000); - try { - dataSource.init(); - } catch (SQLException e) { - throw new RuntimeException(e.getMessage(), e); - } - } - - /** - * 加载远程application.yml配置 - */ - @Override - public void loadRemoteConfig() { - try { - // 加载远程adapter配置 - ConfigItem configItem = getRemoteAdapterConfig(); - if (configItem != null) { - if (configItem.getModifiedTime() != currentConfigTimestamp) { - currentConfigTimestamp = configItem.getModifiedTime(); - overrideLocalCanalConfig(configItem.getContent()); - logger.info("## Loaded remote adapter config: application.yml"); - } - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - /** - * 获取远程application.yml配置 - * - * @return 配置对象 - */ - private ConfigItem getRemoteAdapterConfig() { - String sql = "select name, content, modified_time from canal_config where id=2"; - try (Connection conn = dataSource.getConnection(); - Statement stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery(sql)) { - if (rs.next()) { - ConfigItem configItem = new ConfigItem(); - configItem.setId(2L); - configItem.setName(rs.getString("name")); - configItem.setContent(rs.getString("content")); - configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); - return configItem; - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - return null; - } - - /** - * 覆盖本地application.yml文件 - * - * @param content 文件内容 - */ - private void overrideLocalCanalConfig(String content) { - - try (OutputStreamWriter writer = new OutputStreamWriter( - new FileOutputStream(CommonUtils.getConfPath() + "application.yml"), - StandardCharsets.UTF_8)) { - writer.write(content); - writer.flush(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - /** - * 加载adapter配置 - */ - @Override - public void loadRemoteAdapterConfigs() { - try { - // 加载远程adapter配置 - loadModifiedAdapterConfigs(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - - /** - * 加载有变动的adapter配置 - */ - private void loadModifiedAdapterConfigs() { - Map remoteConfigStatus = new HashMap<>(); - String sql = "select id, category, name, modified_time from canal_adapter_config"; - try (Connection conn = dataSource.getConnection(); - Statement stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery(sql)) { - while (rs.next()) { - ConfigItem configItem = new ConfigItem(); - configItem.setId(rs.getLong("id")); - configItem.setCategory(rs.getString("category")); - configItem.setName(rs.getString("name")); - configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); - remoteConfigStatus.put(configItem.getCategory() + "/" + configItem.getName(), configItem); - } - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - - if (!remoteConfigStatus.isEmpty()) { - List changedIds = new ArrayList<>(); - - for (ConfigItem remoteConfigStat : remoteConfigStatus.values()) { - ConfigItem currentConfig = remoteAdapterConfigs - .get(remoteConfigStat.getCategory() + "/" + remoteConfigStat.getName()); - if (currentConfig == null) { - // 新增 - changedIds.add(remoteConfigStat.getId()); - } else { - // 修改 - if (currentConfig.getModifiedTime() != remoteConfigStat.getModifiedTime()) { - changedIds.add(remoteConfigStat.getId()); - } - } - } - if (!changedIds.isEmpty()) { - String contentsSql = "select id, category, name, content, modified_time from canal_adapter_config where id in (" - + Joiner.on(",").join(changedIds) + ")"; - try (Connection conn = dataSource.getConnection(); - Statement stmt = conn.createStatement(); - ResultSet rs = stmt.executeQuery(contentsSql)) { - while (rs.next()) { - ConfigItem configItemNew = new ConfigItem(); - configItemNew.setId(rs.getLong("id")); - configItemNew.setCategory(rs.getString("category")); - configItemNew.setName(rs.getString("name")); - configItemNew.setContent(rs.getString("content")); - configItemNew.setModifiedTime(rs.getTimestamp("modified_time").getTime()); - - remoteAdapterConfigs.put(configItemNew.getCategory() + "/" + configItemNew.getName(), - configItemNew); - remoteAdapterMonitor.onModify(configItemNew); - } - - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } - } - - for (ConfigItem configItem : remoteAdapterConfigs.values()) { - if (!remoteConfigStatus.containsKey(configItem.getCategory() + "/" + configItem.getName())) { - // 删除 - remoteAdapterConfigs.remove(configItem.getCategory() + "/" + configItem.getName()); - remoteAdapterMonitor.onDelete(configItem.getCategory() + "/" + configItem.getName()); - } - } - } - - /** - * 启动监听数据库变化 - */ - @Override - public void startMonitor() { - // 监听application.yml变化 - executor.scheduleWithFixedDelay(() -> { - try { - loadRemoteConfig(); - } catch (Throwable e) { - logger.error("scan remote application.yml failed", e); - } - }, 10, 3, TimeUnit.SECONDS); - - // 监听adapter变化 - executor.scheduleWithFixedDelay(() -> { - try { - loadRemoteAdapterConfigs(); - } catch (Throwable e) { - logger.error("scan remote adapter configs failed", e); - } - }, 10, 3, TimeUnit.SECONDS); - } - - /** - * 销毁 - */ - @Override - public void destroy() { - executor.shutdownNow(); - try { - dataSource.close(); - } catch (Exception e) { - logger.error(e.getMessage(), e); - } - } -} +package com.alibaba.otter.canal.adapter.launcher.monitor.remote; + +import java.io.FileOutputStream; +import java.io.OutputStreamWriter; +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.alibaba.druid.pool.DruidDataSource; +import com.alibaba.otter.canal.common.utils.CommonUtils; +import com.alibaba.otter.canal.common.utils.NamedThreadFactory; +import com.google.common.base.Joiner; +import com.google.common.collect.MapMaker; + +/** + * 基于数据库的远程配置装载器 + * + * @author rewerma 2019-01-25 下午05:20:16 + * @version 1.0.0 + */ +public class DbRemoteConfigLoader implements RemoteConfigLoader { + + private static final Logger logger = LoggerFactory.getLogger(DbRemoteConfigLoader.class); + + private DruidDataSource dataSource; + + private volatile long currentConfigTimestamp = 0; + private Map remoteAdapterConfigs = new MapMaker().makeMap(); + + private ScheduledExecutorService executor = Executors.newScheduledThreadPool(2, + new NamedThreadFactory("remote-adapter-config-scan")); + + private RemoteAdapterMonitor remoteAdapterMonitor = new RemoteAdapterMonitorImpl(); + + public DbRemoteConfigLoader(String driverName, String jdbcUrl, String jdbcUsername, String jdbcPassword){ + dataSource = new DruidDataSource(); + if (StringUtils.isEmpty(driverName)) { + driverName = "com.mysql.jdbc.Driver"; + } + dataSource.setDriverClassName(driverName); + dataSource.setUrl(jdbcUrl); + dataSource.setUsername(jdbcUsername); + dataSource.setPassword(jdbcPassword); + dataSource.setInitialSize(1); + dataSource.setMinIdle(1); + dataSource.setMaxActive(1); + dataSource.setMaxWait(60000); + dataSource.setTimeBetweenEvictionRunsMillis(60000); + dataSource.setMinEvictableIdleTimeMillis(300000); + try { + dataSource.init(); + } catch (SQLException e) { + throw new RuntimeException(e.getMessage(), e); + } + } + + /** + * 加载远程application.yml配置 + */ + @Override + public void loadRemoteConfig() { + try { + // 加载远程adapter配置 + ConfigItem configItem = getRemoteAdapterConfig(); + if (configItem != null) { + if (configItem.getModifiedTime() != currentConfigTimestamp) { + currentConfigTimestamp = configItem.getModifiedTime(); + overrideLocalCanalConfig(configItem.getContent()); + logger.info("## Loaded remote adapter config: application.yml"); + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 获取远程application.yml配置 + * + * @return 配置对象 + */ + private ConfigItem getRemoteAdapterConfig() { + String sql = "select name, content, modified_time from canal_config where id=2"; + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { + if (rs.next()) { + ConfigItem configItem = new ConfigItem(); + configItem.setId(2L); + configItem.setName(rs.getString("name")); + configItem.setContent(rs.getString("content")); + configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + return configItem; + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + return null; + } + + /** + * 覆盖本地application.yml文件 + * + * @param content 文件内容 + */ + private void overrideLocalCanalConfig(String content) { + + try (OutputStreamWriter writer = new OutputStreamWriter( + new FileOutputStream(CommonUtils.getConfPath() + "application.yml"), + StandardCharsets.UTF_8)) { + writer.write(content); + writer.flush(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 加载adapter配置 + */ + @Override + public void loadRemoteAdapterConfigs() { + try { + // 加载远程adapter配置 + loadModifiedAdapterConfigs(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + /** + * 加载有变动的adapter配置 + */ + private void loadModifiedAdapterConfigs() { + Map remoteConfigStatus = new HashMap<>(); + String sql = "select id, category, name, modified_time from canal_adapter_config"; + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { + while (rs.next()) { + ConfigItem configItem = new ConfigItem(); + configItem.setId(rs.getLong("id")); + configItem.setCategory(rs.getString("category")); + configItem.setName(rs.getString("name")); + configItem.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + remoteConfigStatus.put(configItem.getCategory() + "/" + configItem.getName(), configItem); + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + + if (!remoteConfigStatus.isEmpty()) { + List changedIds = new ArrayList<>(); + + for (ConfigItem remoteConfigStat : remoteConfigStatus.values()) { + ConfigItem currentConfig = remoteAdapterConfigs + .get(remoteConfigStat.getCategory() + "/" + remoteConfigStat.getName()); + if (currentConfig == null) { + // 新增 + changedIds.add(remoteConfigStat.getId()); + } else { + // 修改 + if (currentConfig.getModifiedTime() != remoteConfigStat.getModifiedTime()) { + changedIds.add(remoteConfigStat.getId()); + } + } + } + if (!changedIds.isEmpty()) { + String contentsSql = "select id, category, name, content, modified_time from canal_adapter_config where id in (" + + Joiner.on(",").join(changedIds) + ")"; + try (Connection conn = dataSource.getConnection(); + Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery(contentsSql)) { + while (rs.next()) { + ConfigItem configItemNew = new ConfigItem(); + configItemNew.setId(rs.getLong("id")); + configItemNew.setCategory(rs.getString("category")); + configItemNew.setName(rs.getString("name")); + configItemNew.setContent(rs.getString("content")); + configItemNew.setModifiedTime(rs.getTimestamp("modified_time").getTime()); + + remoteAdapterConfigs.put(configItemNew.getCategory() + "/" + configItemNew.getName(), + configItemNew); + remoteAdapterMonitor.onModify(configItemNew); + } + + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + } + + for (ConfigItem configItem : remoteAdapterConfigs.values()) { + if (!remoteConfigStatus.containsKey(configItem.getCategory() + "/" + configItem.getName())) { + // 删除 + remoteAdapterConfigs.remove(configItem.getCategory() + "/" + configItem.getName()); + remoteAdapterMonitor.onDelete(configItem.getCategory() + "/" + configItem.getName()); + } + } + } + + /** + * 启动监听数据库变化 + */ + @Override + public void startMonitor() { + // 监听application.yml变化 + executor.scheduleWithFixedDelay(() -> { + try { + loadRemoteConfig(); + } catch (Throwable e) { + logger.error("scan remote application.yml failed", e); + } + }, 10, 3, TimeUnit.SECONDS); + + // 监听adapter变化 + executor.scheduleWithFixedDelay(() -> { + try { + loadRemoteAdapterConfigs(); + } catch (Throwable e) { + logger.error("scan remote adapter configs failed", e); + } + }, 10, 3, TimeUnit.SECONDS); + } + + /** + * 销毁 + */ + @Override + public void destroy() { + executor.shutdownNow(); + try { + dataSource.close(); + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } +} diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 7c455118..38020409 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -7,10 +7,10 @@ spring: default-property-inclusion: non_null canal.conf: - mode: kafka # kafka rocketMQ -# canalServerHost: 127.0.0.1:11111 + mode: tcp # kafka rocketMQ + canalServerHost: 127.0.0.1:11111 # zookeeperHosts: slave1:2181 - mqServers: 127.0.0.1:9092 #or rocketmq +# mqServers: 127.0.0.1:9092 #or rocketmq # flatMessage: true batchSize: 500 syncBatchSize: 1000 @@ -58,6 +58,9 @@ canal.conf: # hbase.zookeeper.property.clientPort: 2181 # zookeeper.znode.parent: /hbase # - name: es -# hosts: 127.0.0.1:9300 +# hosts: 127.0.0.1:9300 # 127.0.0.1:9200 for rest mode # properties: +# mode: transport # or rest +# # security.auth: test:123456 # only used for rest mode # cluster.name: elasticsearch + diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java index a3d8c7f0..704b03a8 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java @@ -177,6 +177,7 @@ public class CanalKafkaProducer implements CanalMQProducer { // 每条记录需要flush produce(topicName, records, true); + records.clear(); } } }