...
- we use the following configuration (see org.openhubframework.openhub.core.common.asynch.AsynchMessageRoute#URI_ASYNC_PROCESSING_MSG)
- ASYNCH_CONCURRENT_CONSUMERS corresponds to the parameter ohf.asynch.concurrentConsumers, see OpenHub configuration
- PRIORITY_QUEUE_FACTORY refers to Spring bean with the name "priorityQueueFactory"
- queue size must be big enough for saving incoming requests otherwise thread that adds new message will wait
| Code Block |
|---|
public static final String URI_ASYNC_PROCESSING_MSG = "seda:asynch_message_route"
+ "?concurrentConsumers={{" + ASYNCH_CONCURRENT_CONSUMERS + "}}&waitForTaskToComplete=Never"
+ "&blockWhenFull=true&queueFactory=#" + PRIORITY_QUEUE_FACTORY; |
...
| Code Block |
|---|
2019-01-14 19:52:48 [Camel (camelContext) thread #20 - seda://asynch_message_route, ALTA, ID:95fc3756-182d-11e9-b9c2-005056010488] WARN syncProcessOut_route - Message (msg_id = 37357, correlationId = ID:95fc3756-182d-11e9-b9c2-005056010488) was obsolete, stopped further processing. |
Thread pools
...
- see threading model in Apache Camel
- default ThreadPoolProfile is as follows (defined in org.apache.camel.impl.DefaultExecutorServiceManager):
- pool size - threads to keep minimum in pool
- max pool size - the maximum pool size
- max queue size - the maximum number of tasks in the work queue
| Code Block |
|---|
defaultProfile = new ThreadPoolProfile(defaultThreadPoolProfileId);
defaultProfile.setDefaultProfile(true);
defaultProfile.setPoolSize(10);
defaultProfile.setMaxPoolSize(20);
defaultProfile.setKeepAliveTime(60L);
defaultProfile.setTimeUnit(TimeUnit.SECONDS);
defaultProfile.setMaxQueueSize(1000);
defaultProfile.setAllowCoreThreadTimeOut(false);
defaultProfile.setRejectedPolicy(ThreadPoolRejectedPolicy.CallerRuns); |
- we have small changes in our custom configuration (see org.openhubframework.openhub.core.config.CamelConfig)
- MAX_THREAD_POOL_SIZE = 30
| Code Block |
|---|
ThreadPoolProfile threadPoolProfile = camelContext.getExecutorServiceManager().getDefaultThreadPoolProfile();
threadPoolProfile.setId(DEFAULT_THREAD_PROFILE);
threadPoolProfile.setMaxPoolSize(MAX_THREAD_POOL_SIZE); |
Throttling
Throttling can be disabled: ohf.disable.throttling = false
Caching
Use Cache / memory-grid functionality to increase performance when often data reading is necessary.