Single consumer for ordered queue in distributed system

[taalexlistex]

Hello. I want my messages to be processed in the same order as they were added to the queue. I can set concurrency=1 in RqueueListener annotation and it works if one application is run. But is there any option to use only one consumer across multiple instances? Is it possible to use distributed lock and pop messages only from one machine? thanks

[taalexlistex]

at the moment implemented by overriding pop method in RqueueMessageTemplate

@Bean
    public RqueueMessageTemplate rqueueMessageTemplate(RqueueConfig config, Lock lock) {
        return new RqueueMessageTemplateImpl(config.getConnectionFactory(), config.getReactiveRedisConnectionFactory()) {
            @Override
            public List<RqueueMessage> pop(String queueName, String processingQueueName, String processingChannelName, long visibilityTimeout, int count) {
                String token = null;
                if (queueName.contains(BILLING_QUEUE)) {
                    token = lock.acquire(List.of(queueName), DISTRIBUTED_QUEUE_LOCK, 60000);
                    if (token == null) {
                        return new ArrayList<>();
                    }
                }
                List<RqueueMessage> messages = super.pop(queueName, processingQueueName, processingChannelName, visibilityTimeout, count);
                if (token != null) {
                    lock.release(List.of(queueName), DISTRIBUTED_QUEUE_LOCK, token);
                }
                return messages;
            }
        };
    }

[taalexlistex]

also can see huge problem that one message cannot be handled faster than 200ms
my handler is empty at the moment so 2 seconds for 10 messages in unappropriated time
poll() has to be executed right after all tasks are done if time < pollingInterval

void poll() {
    if (!hasAvailableThreads(queueDetail, queueThreadPool)) {
      log(Level.WARN, "No Threads are available sleeping {}Ms", null, pollingInterval);
      TimeoutUtils.sleepLog(pollingInterval, false);
    } else {
      super.poll(-1, queueDetail.getName(), queueDetail, queueThreadPool);
    }
  }
---logs---
2023-03-02 17:17:42.336  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:42.547  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:42.778  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:42.996  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:43.231  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:43.442  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:43.662  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:43.887  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:44.095  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms
2023-03-02 17:17:44.317  WARN   --- [enerContainer-1] c.g.s.r.listener.RqueueMessagePoller     : [billing-queue] No Threads are available sleeping 200Ms

[sonus21]

You can set the polling interval to zero, currently, there's no way to know if all the tasks have been completed, due to the shared thread pool.

[taalexlistex]

Yeah, thanks. It would be nice to have pollingInterval parameter in @MessageListener and copy it to a poller if set. So I could only reduce interval for a specific queue. Thank you.

[sonus21]

@taalexlistex can you tell me if you're using version 3.x? I can make that small change quickly.

[taalexlistex]

2.x, would be nice. thanks a lot