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

added env variable to regulate pubsud queue executor threads pool size

Browse Source
This commit is contained in:
dashevchenko 2023-11-13 13:14:37 +02:00
parent afa54ef277
commit 442fe9029e
9 changed files with 26 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
application/src/main/resources/thingsboard.yml
View File

@ -1411,6 +1411,8 @@ queue:
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
# Number of messages per consumer
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
# Number of threads of pubsub executor provider
executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_SIZE:50}"
queue-properties:
# Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

10
common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubProducerTemplate.java
View File

@ -18,6 +18,7 @@ package org.thingsboard.server.queue.pubsub;
import com.google.api.core.ApiFuture;
import com.google.api.core.ApiFutureCallback;
import com.google.api.core.ApiFutures;
import com.google.api.gax.core.InstantiatingExecutorProvider;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.gson.Gson;
import com.google.protobuf.ByteString;
@ -120,7 +121,14 @@ public class TbPubSubProducerTemplate<T extends TbQueueMsg> implements TbQueuePr
try {
admin.createTopicIfNotExists(topic);
ProjectTopicName topicName = ProjectTopicName.of(pubSubSettings.getProjectId(), topic);
Publisher publisher = Publisher.newBuilder(topicName).setCredentialsProvider(pubSubSettings.getCredentialsProvider()).build();
InstantiatingExecutorProvider executorProvider =
InstantiatingExecutorProvider.newBuilder()
.setExecutorThreadCount(pubSubSettings.getExecutorThreadPoolSize())
.build();
Publisher publisher = Publisher.newBuilder(topicName)
.setCredentialsProvider(pubSubSettings.getCredentialsProvider())
.setExecutorProvider(executorProvider)
.build();
publisherMap.put(topic, publisher);
return publisher;
} catch (IOException e) {

3
common/queue/src/main/java/org/thingsboard/server/queue/pubsub/TbPubSubSettings.java
View File

@ -46,6 +46,9 @@ public class TbPubSubSettings {
@Value("${queue.pubsub.max_messages}")
private int maxMessages;
@Value("${queue.pubsub.executor_thread_pool_size}")
private int executorThreadPoolSize;
private CredentialsProvider credentialsProvider;
@PostConstruct

2
msa/vc-executor/src/main/resources/tb-vc-executor.yml
View File

@ -172,6 +172,8 @@ queue:
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}" #in bytes
# Number of messages per consumer
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
# Number of threads of pubsub executor provider
executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_SIZE:50}"
queue-properties:
# Pub/Sub properties for Core subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
core: "${TB_QUEUE_PUBSUB_CORE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

2
transport/coap/src/main/resources/tb-coap-transport.yml
View File

@ -295,6 +295,8 @@ queue:
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
# Number of messages per consumer
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
# Number of threads of pubsub executor provider
executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_SIZE:50}"
queue-properties:
# Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

2
transport/http/src/main/resources/tb-http-transport.yml
View File

@ -278,6 +278,8 @@ queue:
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
# Number of messages per a consumer
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
# Number of threads of pubsub executor provider
executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_SIZE:50}"
queue-properties:
# Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consume again
rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

2
transport/lwm2m/src/main/resources/tb-lwm2m-transport.yml
View File

@ -374,6 +374,8 @@ queue:
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
# Number of messages per consumer
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
# Number of threads of pubsub executor provider
executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_SIZE:50}"
queue-properties:
# Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

2
transport/mqtt/src/main/resources/tb-mqtt-transport.yml
View File

@ -311,6 +311,8 @@ queue:
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
# Number of messages per consumer
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
# Number of threads of pubsub executor provider
executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_SIZE:50}"
queue-properties:
# Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"

2
transport/snmp/src/main/resources/tb-snmp-transport.yml
View File

@ -264,6 +264,8 @@ queue:
max_msg_size: "${TB_QUEUE_PUBSUB_MAX_MSG_SIZE:1048576}"
# Number of messages per consumer
max_messages: "${TB_QUEUE_PUBSUB_MAX_MESSAGES:1000}"
# Number of threads of pubsub executor provider
executor_thread_pool_size: "${TB_QUEUE_PUBSUB_EXECUTOR_THREAD_SIZE:50}"
queue-properties:
# Pub/Sub properties for Rule Engine subscribers, messages which will commit after ackDeadlineInSec period can be consumed again
rule-engine: "${TB_QUEUE_PUBSUB_RE_QUEUE_PROPERTIES:ackDeadlineInSec:30;messageRetentionInSec:604800}"