dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed concurrency issues on high load

Browse Source
This commit is contained in:
Andrew Shvayka 2018-11-15 10:14:14 +02:00
parent 8fd37663b9
commit eb47eb3c4f
3 changed files with 7 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
application/src/main/java/org/thingsboard/server/service/transport/RemoteTransportApiService.java
View File

@ -29,6 +29,7 @@ import org.thingsboard.server.kafka.*;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy; import javax.annotation.PreDestroy;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.SynchronousQueue; import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
@ -67,7 +68,7 @@ public class RemoteTransportApiService {
@PostConstruct @PostConstruct
public void init() { public void init() {
this.transportCallbackExecutor = new ThreadPoolExecutor(0, 100, 60L, TimeUnit.SECONDS, new SynchronousQueue<>()); this.transportCallbackExecutor = new ThreadPoolExecutor(0, 100, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaProducerTemplate.builder(); TBKafkaProducerTemplate.TBKafkaProducerTemplateBuilder<TransportApiResponseMsg> responseBuilder = TBKafkaProducerTemplate.builder();
responseBuilder.settings(kafkaSettings); responseBuilder.settings(kafkaSettings);

3
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/AbstractTransportService.java
View File

@ -32,6 +32,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.SynchronousQueue; import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
@ -277,7 +278,7 @@ public abstract class AbstractTransportService implements TransportService {
new TbRateLimits(perDevicesLimitsConf); new TbRateLimits(perDevicesLimitsConf);
} }
this.schedulerExecutor = Executors.newSingleThreadScheduledExecutor(); this.schedulerExecutor = Executors.newSingleThreadScheduledExecutor();
this.transportCallbackExecutor = new ThreadPoolExecutor(0, 20, 60L, TimeUnit.SECONDS, new SynchronousQueue<>()); this.transportCallbackExecutor = new ThreadPoolExecutor(0, 20, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
this.schedulerExecutor.scheduleAtFixedRate(this::checkInactivityAndReportActivity, sessionReportTimeout, sessionReportTimeout, TimeUnit.MILLISECONDS); this.schedulerExecutor.scheduleAtFixedRate(this::checkInactivityAndReportActivity, sessionReportTimeout, sessionReportTimeout, TimeUnit.MILLISECONDS);
} }

4
dao/src/main/java/org/thingsboard/server/dao/util/AbstractBufferedRateExecutor.java
View File

@ -31,7 +31,9 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque; import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException; import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
@ -70,7 +72,7 @@ public abstract class AbstractBufferedRateExecutor<T extends AsyncTask, F extend
this.concurrencyLimit = concurrencyLimit; this.concurrencyLimit = concurrencyLimit;
this.queue = new LinkedBlockingDeque<>(queueLimit); this.queue = new LinkedBlockingDeque<>(queueLimit);
this.dispatcherExecutor = Executors.newFixedThreadPool(dispatcherThreads); this.dispatcherExecutor = Executors.newFixedThreadPool(dispatcherThreads);
this.callbackExecutor = Executors.newFixedThreadPool(callbackThreads); this.callbackExecutor = new ThreadPoolExecutor(callbackThreads, 50, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
this.timeoutExecutor = Executors.newSingleThreadScheduledExecutor(); this.timeoutExecutor = Executors.newSingleThreadScheduledExecutor();
this.perTenantLimitsEnabled = perTenantLimitsEnabled; this.perTenantLimitsEnabled = perTenantLimitsEnabled;
this.perTenantLimitsConfiguration = perTenantLimitsConfiguration; this.perTenantLimitsConfiguration = perTenantLimitsConfiguration;