This is an automated email from the ASF dual-hosted git repository. ningjiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 690ebe1916fdcfa907866c1b8896ee17d8a52113 Author: Lei Zhang <[email protected]> AuthorDate: Fri Jul 26 14:58:30 2019 +0800 SCB-1369 Use synchronization methods to avoid conflicts between thread refreshTimer and batch writes --- .../ElasticsearchTransactionRepository.java | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java index 601249e..b7dc0a1 100644 --- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java +++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java @@ -42,6 +42,7 @@ public class ElasticsearchTransactionRepository implements TransactionRepository private int batchSizeCounter; private int refreshTime; private final List<IndexQuery> queries = new ArrayList<>(); + private final Boolean lock = true; public ElasticsearchTransactionRepository( ElasticsearchTemplate template, MetricsService metricsService, int batchSize, @@ -50,10 +51,12 @@ public class ElasticsearchTransactionRepository implements TransactionRepository this.metricsService = metricsService; this.batchSize = batchSize; this.refreshTime = refreshTime; - if (this.refreshTime > 0) { new Thread(new RefreshTimer(), "elasticsearch-repository-refresh").start(); } + if(!this.template.indexExists(INDEX_NAME)){ + this.template.createIndex(INDEX_NAME); + } } @Override @@ -63,10 +66,11 @@ public class ElasticsearchTransactionRepository implements TransactionRepository batchSizeCounter++; metricsService.metrics().doRepositoryReceived(); if (batchSize == 0 || batchSizeCounter == batchSize) { - save(begin); - batchSizeCounter = 0; - queries.clear(); - + synchronized (lock){ + save(begin); + batchSizeCounter = 0; + queries.clear(); + } } } @@ -93,8 +97,10 @@ public class ElasticsearchTransactionRepository implements TransactionRepository public void run() { while (true) { try { - if (!queries.isEmpty()) { - save(System.currentTimeMillis()); + synchronized (lock){ + if (!queries.isEmpty()) { + save(System.currentTimeMillis()); + } } } catch (Exception e) { LOG.error(e.getMessage(), e);
