Rqueue Documentation

The default GenericMessageConverter supports single-level generic envelope types like Event<T>. The type parameter T is resolved at serialization by inspecting the runtime class of the field value.

Constraints:

• T must be a non-generic concrete class (e.g., Order, not List<Order>).
• At least one field of type T on the envelope must be non-null during serialization.
• Multi-level nesting (e.g., Wrapper<Event<T>>) is not supported.
• List<T> is supported only if T is a non-generic concrete class.

// A generic envelope type
public class Event<T> {
  private String id;
  private T payload;
  // getters/setters ...
}

// Enqueue
Event<Order> event = new Event<>("evt-123", order);
rqueueMessageEnqueuer.enqueue("order-queue", event);

// Consume
@RqueueListener(value = "order-queue")
public void onEvent(Event<Order> event) { ... }

// The serialized form encodes both the envelope class and the type parameter:
// {"msg":"...","name":"com.example.Event#com.example.Order"}

Rqueue manages concurrency through task executors and queue-level settings:

• Shared Executor: By default, the number of task executors is twice the number of queues. You can provide a custom ThreadPoolTaskExecutor using factory.setTaskExecutor(executor).
• Queue-level Concurrency: Use the @RqueueListener annotation's concurrency field.
  • A fixed number (e.g., concurrency = 10) creates a dedicated task executor for that queue.
  • A range (e.g., concurrency = "5-10") also uses a dedicated executor.
  • If omitted, the queue uses the shared task executor.
• Global Worker Limit: Use factory.setMaxNumWorkers(int) to set a global limit on workers.
• Batch Size: The @RqueueListener batchSize field determines how many messages are fetched at once. Default is 10 for listeners with explicit concurrency, and 1 for others.

Important Note on queueCapacity: A non-zero queueCapacity in your executor can lead to duplicate message processing if a message sits in the executor's queue longer than its visibilityTimeout. Ensure visibilityTimeout is long enough to accommodate queuing delays.

class RqueueConfiguration {
  @Bean
  public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory() {
    SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
    //...
    factory.setMaxNumWorkers(10);
    return factory;
  }
}

class RqueueConfiguration {

  @Bean
  public SimpleRqueueListenerContainerFactory simpleRqueueListenerContainerFactory() {
    SimpleRqueueListenerContainerFactory factory = new SimpleRqueueListenerContainerFactory();
    //...
    ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
    threadPoolTaskExecutor.setThreadNamePrefix("taskExecutor");
    threadPoolTaskExecutor.setCorePoolSize(10);
    threadTaskExecutor.setMaxPoolSize(50);
    threadPoolTaskExecutor.setQueueCapacity(0);
    threadPoolTaskExecutor.afterPropertiesSet();
    factory.setTaskExecutor(threadPoolTaskExecutor);
    return factory;
  }
}

Rqueue uses a shared set of JetStream KV buckets for all queues. Scoping is handled via key prefixes within the buckets.

Core Buckets:

Important: TTL is fixed at the moment of bucket creation. Changing Rqueue properties will not update an existing bucket's TTL.

When using the NATS backend (via rqueue-nats), certain features available in the Redis backend are not supported due to the architectural differences of NATS JetStream.

Supported Operations:

• pauseUnpauseQueue via NatsRqueueUtilityService
• getDataType
• aggregateDataCounter
• soft deleteMessage

Unsupported Operations (will throw BackendCapabilityException or UnsupportedOperationException):

• Message Manipulation: moveMessage, enqueueMessage, and makeEmpty are not supported as JetStream does not provide equivalent primitives.
• Scheduling: Delayed, scheduled, or cron messages are not supported.
• Advanced Queueing: Cross-queue priorityGroup weighting is not honored (will trigger a boot warning).
• Concurrency: Elastic @RqueueListener.concurrency (where min < max) falls back to a fixed max concurrency.
• Handler Configuration: @RqueueHandler(primary) is ignored (will trigger a boot warning).