How can I use ArrayBlockingQueue with Spring Boot? I have configured queue as below -

@Configuration
public class MessageQueueConfig {

 @Bean
 public ArrayBlockingQueue arrayBlockingQueue() {
    ArrayBlockingQueue<Message> arrayBlockingQueue = new ArrayBlockingQueue<>(50000);
    return arrayBlockingQueue;
 }
}

On the api call I am offering data to queue

@Autowired
private MessageQueueConfig queueConfig;

...
queueConfig.arrayBlockingQueue().offer(message, 5, TimeUnit.SECONDS)

To poll the message from queue, do I need to use threading? or how can I poll a message from queue? Can directly use @Autowired for queue in consumer and poll the message in consumer

CodePudding user response：

To poll the message from queue, do I need to use threading?

According to documentation, BlockingQueue implementations are thread-safe. You can use threads if needed.

Can directly use @Autowired for queue in consumer and poll the message in consumer

In the Consumer do something like this:

@Autowired
private BlockingQueue<Message> arrayBlockingQueue;

//...
arrayBlockingQueue.poll(5, TimeUnit.SECONDS)

CodePudding user response：

The direct answer is No.

ArrayBlockingQueue is an implementation of the BlockingQueue interface which have locks implemented.

For reference, i am posting below source code implementation for poll() and take() methods of ArrayBlockingQueue.

     public E poll() {
    final ReentrantLock lock = this.lock;
     lock.lock();
     try {
         if (count == 0)
             return null;
         E x = extract();
         return x;
     } finally {
         lock.unlock();
     }
}

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
       try {
            while (count == 0)
                notEmpty.await();
        } catch (InterruptedException ie) {
             notEmpty.signal(); // propagate to non-interrupted thread
            throw ie;
        }
        E x = extract();
         return x;
    } finally {
        lock.unlock();
     }
}