简柏林 / thingskit · Commits

Commit c2596b8ecbfb3c935164096f9a4a59b8c0411a74

Authored by YevhenBondarenko
Committed by Andrew Shvayka
1 parent f5b4ebbd

added SequentialByOriginator queue to coap, http, mqtt and refactored

Inline Side-by-side
Showing 5 changed files with 65 additions and 20 deletions
... ... @@ -48,7 +48,7 @@ public class TbSynchronizationBeginNode implements TbNode {
48 48
49 49 @Override
50 50 public void onMsg(TbContext ctx, TbMsg msg) {
51   - log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_WITHIN_ORIGINATOR instead.");
  51 + log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_BY_ORIGINATOR instead.");
52 52 ctx.tellSuccess(msg);
53 53 }
54 54
... ...
... ... @@ -49,7 +49,7 @@ public class TbSynchronizationEndNode implements TbNode {
49 49
50 50 @Override
51 51 public void onMsg(TbContext ctx, TbMsg msg) {
52   - log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_WITHIN_ORIGINATOR instead.");
  52 + log.warn("Synchronization Start/End nodes are deprecated since TB 2.5. Use queue with submit strategy SEQUENTIAL_BY_ORIGINATOR instead.");
53 53 ctx.tellSuccess(msg);
54 54 }
55 55
... ...
... ... @@ -155,19 +155,19 @@ queue:
155 155 pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}"
156 156 stats:
157 157 enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:true}"
158   - print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}"
  158 + print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:60000}"
159 159 queues:
160   - - name: "Main"
  160 + - name: "${TB_QUEUE_RE_MAIN_QUEUE_NAME:Main}"
161 161 topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}"
162 162 poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}"
163 163 partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}"
164 164 pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}"
165 165 submit-strategy:
166   - type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
  166 + type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
167 167 # For BATCH only
168 168 batch-size: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_BATCH_SIZE:1000}" # Maximum number of messages in batch
169 169 processing-strategy:
170   - type: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  170 + type: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_TYPE:SKIP_ALL_FAILURES}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
171 171 # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
172 172 retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
173 173 failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
... ... @@ -175,10 +175,10 @@ queue:
175 175 - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}"
176 176 topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}"
177 177 poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}"
178   - partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}"
  178 + partitions: "${TB_QUEUE_RE_HP_PARTITIONS:10}"
179 179 pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}"
180 180 submit-strategy:
181   - type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_WITHIN_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
  181 + type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
182 182 # For BATCH only
183 183 batch-size: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
184 184 processing-strategy:
... ... @@ -187,6 +187,21 @@ queue:
187 187 retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited
188 188 failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
189 189 pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
  190 + - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}"
  191 + topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}"
  192 + poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}"
  193 + partitions: "${TB_QUEUE_RE_SQ_PARTITIONS:10}"
  194 + pack-processing-timeout: "${TB_QUEUE_RE_SQ_PACK_PROCESSING_TIMEOUT_MS:60000}"
  195 + submit-strategy:
  196 + type: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_BY_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
  197 + # For BATCH only
  198 + batch-size: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
  199 + processing-strategy:
  200 + type: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  201 + # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  202 + retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
  203 + failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
  204 + pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
190 205 transport:
191 206 # For high priority notifications that require minimum latency and processing time
192 207 notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}"
... ...
... ... @@ -156,19 +156,19 @@ queue:
156 156 pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}"
157 157 stats:
158 158 enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:true}"
159   - print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}"
  159 + print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:60000}"
160 160 queues:
161   - - name: "Main"
  161 + - name: "${TB_QUEUE_RE_MAIN_QUEUE_NAME:Main}"
162 162 topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}"
163 163 poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}"
164 164 partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}"
165 165 pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}"
166 166 submit-strategy:
167   - type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
  167 + type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
168 168 # For BATCH only
169 169 batch-size: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_BATCH_SIZE:1000}" # Maximum number of messages in batch
170 170 processing-strategy:
171   - type: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  171 + type: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_TYPE:SKIP_ALL_FAILURES}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
172 172 # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
173 173 retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
174 174 failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
... ... @@ -176,10 +176,10 @@ queue:
176 176 - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}"
177 177 topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}"
178 178 poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}"
179   - partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}"
  179 + partitions: "${TB_QUEUE_RE_HP_PARTITIONS:10}"
180 180 pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}"
181 181 submit-strategy:
182   - type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_WITHIN_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
  182 + type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
183 183 # For BATCH only
184 184 batch-size: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
185 185 processing-strategy:
... ... @@ -188,6 +188,21 @@ queue:
188 188 retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited
189 189 failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
190 190 pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
  191 + - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}"
  192 + topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}"
  193 + poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}"
  194 + partitions: "${TB_QUEUE_RE_SQ_PARTITIONS:10}"
  195 + pack-processing-timeout: "${TB_QUEUE_RE_SQ_PACK_PROCESSING_TIMEOUT_MS:60000}"
  196 + submit-strategy:
  197 + type: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_BY_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
  198 + # For BATCH only
  199 + batch-size: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
  200 + processing-strategy:
  201 + type: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  202 + # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  203 + retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
  204 + failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
  205 + pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
191 206 transport:
192 207 # For high priority notifications that require minimum latency and processing time
193 208 notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}"
... ...
... ... @@ -176,19 +176,19 @@ queue:
176 176 pack-processing-timeout: "${TB_QUEUE_RULE_ENGINE_PACK_PROCESSING_TIMEOUT_MS:60000}"
177 177 stats:
178 178 enabled: "${TB_QUEUE_RULE_ENGINE_STATS_ENABLED:true}"
179   - print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:10000}"
  179 + print-interval-ms: "${TB_QUEUE_RULE_ENGINE_STATS_PRINT_INTERVAL_MS:60000}"
180 180 queues:
181   - - name: "Main"
  181 + - name: "${TB_QUEUE_RE_MAIN_QUEUE_NAME:Main}"
182 182 topic: "${TB_QUEUE_RE_MAIN_TOPIC:tb_rule_engine.main}"
183 183 poll-interval: "${TB_QUEUE_RE_MAIN_POLL_INTERVAL_MS:25}"
184 184 partitions: "${TB_QUEUE_RE_MAIN_PARTITIONS:10}"
185 185 pack-processing-timeout: "${TB_QUEUE_RE_MAIN_PACK_PROCESSING_TIMEOUT_MS:60000}"
186 186 submit-strategy:
187   - type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
  187 + type: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
188 188 # For BATCH only
189 189 batch-size: "${TB_QUEUE_RE_MAIN_SUBMIT_STRATEGY_BATCH_SIZE:1000}" # Maximum number of messages in batch
190 190 processing-strategy:
191   - type: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  191 + type: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_TYPE:SKIP_ALL_FAILURES}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
192 192 # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
193 193 retries: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
194 194 failure-percentage: "${TB_QUEUE_RE_MAIN_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
... ... @@ -196,10 +196,10 @@ queue:
196 196 - name: "${TB_QUEUE_RE_HP_QUEUE_NAME:HighPriority}"
197 197 topic: "${TB_QUEUE_RE_HP_TOPIC:tb_rule_engine.hp}"
198 198 poll-interval: "${TB_QUEUE_RE_HP_POLL_INTERVAL_MS:25}"
199   - partitions: "${TB_QUEUE_RE_HP_PARTITIONS:3}"
  199 + partitions: "${TB_QUEUE_RE_HP_PARTITIONS:10}"
200 200 pack-processing-timeout: "${TB_QUEUE_RE_HP_PACK_PROCESSING_TIMEOUT_MS:60000}"
201 201 submit-strategy:
202   - type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_WITHIN_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_WITHIN_ORIGINATOR, SEQUENTIAL_WITHIN_TENANT, SEQUENTIAL
  202 + type: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_TYPE:BURST}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
203 203 # For BATCH only
204 204 batch-size: "${TB_QUEUE_RE_HP_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
205 205 processing-strategy:
... ... @@ -208,6 +208,21 @@ queue:
208 208 retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRIES:0}" # Number of retries, 0 is unlimited
209 209 failure-percentage: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
210 210 pause-between-retries: "${TB_QUEUE_RE_HP_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
  211 + - name: "${TB_QUEUE_RE_SQ_QUEUE_NAME:SequentialByOriginator}"
  212 + topic: "${TB_QUEUE_RE_SQ_TOPIC:tb_rule_engine.sq}"
  213 + poll-interval: "${TB_QUEUE_RE_SQ_POLL_INTERVAL_MS:25}"
  214 + partitions: "${TB_QUEUE_RE_SQ_PARTITIONS:10}"
  215 + pack-processing-timeout: "${TB_QUEUE_RE_SQ_PACK_PROCESSING_TIMEOUT_MS:60000}"
  216 + submit-strategy:
  217 + type: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_TYPE:SEQUENTIAL_BY_ORIGINATOR}" # BURST, BATCH, SEQUENTIAL_BY_ORIGINATOR, SEQUENTIAL_BY_TENANT, SEQUENTIAL
  218 + # For BATCH only
  219 + batch-size: "${TB_QUEUE_RE_SQ_SUBMIT_STRATEGY_BATCH_SIZE:100}" # Maximum number of messages in batch
  220 + processing-strategy:
  221 + type: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_TYPE:RETRY_FAILED_AND_TIMED_OUT}" # SKIP_ALL_FAILURES, RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  222 + # For RETRY_ALL, RETRY_FAILED, RETRY_TIMED_OUT, RETRY_FAILED_AND_TIMED_OUT
  223 + retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRIES:3}" # Number of retries, 0 is unlimited
  224 + failure-percentage: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_FAILURE_PERCENTAGE:0}" # Skip retry if failures or timeouts are less then X percentage of messages;
  225 + pause-between-retries: "${TB_QUEUE_RE_SQ_PROCESSING_STRATEGY_RETRY_PAUSE:5}"# Time in seconds to wait in consumer thread before retries;
211 226 transport:
212 227 # For high priority notifications that require minimum latency and processing time
213 228 notifications_topic: "${TB_QUEUE_TRANSPORT_NOTIFICATIONS_TOPIC:tb_transport.notifications}"
... ...