java BlockingQueue没有阻塞查看？

Question

我有一个对象阻塞队列。

我想编写一个线程，该线程会阻塞直到队列中有一个对象。与BlockingQueue.take()提供的功能类似。

但是，由于我不知道是否能够成功处理该对象，因此我只想 peek() 而不是删除该对象。仅当我能够成功处理该对象时，我才想删除该对象。

所以，我想要一个阻塞 peek() 函数。目前，根据 javadoc，peek() 仅在队列为空时返回。

我错过了什么吗？还有其他方法可以实现此功能吗？

编辑：

如果我只是使用线程安全队列并偷看并睡觉，有什么想法吗？

public void run() {
    while (!exit) {
        while (queue.size() != 0) {
            Object o =  queue.peek();
            if (o != null) {
                if (consume(o) == true) {
                    queue.remove();
                } else {
                    Thread.sleep(10000); //need to backoff (60s) and try again
                }
            }
        }
        Thread.sleep(1000); //wait 1s for object on queue
    }
}

请注意，我只有一个消费者线程和一个（单独的）生产者线程。我想这不如使用 BlockingQueue 有效......任何评论表示赞赏。

Answer 1

您可以使用 LinkedBlockingDeque 并在物理上从队列中删除该项目（使用 takeLast()），但如果使用 putLast(E e) 处理失败，则在队列末尾再次替换该项目代码>.同时，您的“生产者”将使用 putFirst(E e) 将元素添加到队列的前面。

您始终可以将此行为封装在您自己的 Queue 实现中，并提供一个 blockingPeek() 方法来执行 takeLast() ，然后执行 putLast () 在底层 LinkedBlockingDeque 的幕后。因此，从调用客户端的角度来看，该元素永远不会从队列中删除。

Answer 2

但是，由于我不知道是否能够成功处理该对象，因此我只想 peek() 而不是删除该对象。仅当我能够成功处理该对象时，我才想删除该对象。

一般来说，它不是线程安全的。如果在您 peek() 并确定可以成功处理该对象之后，但在 take() 删除并处理该对象之前，另一个线程获取该对象，该怎么办？

Answer 3

您是否还可以将事件监听器队列添加到阻塞队列，然后当将某些内容添加到（阻塞）队列时，将事件发送给您的监听器？您可以让线程阻塞，直到调用它的 actionPerformed 方法。

Answer 4

我唯一知道的是 Apache Commons Collections 中的 nofollow noreferrer">BlockingBuffer：

如果调用 get 或 remove
一个空的 Buffer，调用线程
等待添加或添加的通知
addAll操作已完成。

get() 相当于 peek()，并且可以通过修饰使 Buffer 表现得像 BlockingQueue一个 UnboundedFifoBuffer 阻塞缓冲区

Answer 5

本身不是答案，但是： JDK-6653412 声称这不是有效的用例。

Answer 6

简单的回答是，实际上没有办法进行阻塞查看，除非您自己使用阻塞查看（）来实现阻塞队列。

我错过了什么吗？

peek() 可能会带来并发问题 -

• 如果您无法处理您的 peek() 消息 - 它将被留在队列中，除非您有多个消费者。
• 如果您无法处理该对象，谁将把该对象从队列中取出？
• 如果您有多个消费者，您会在 peek() 和另一个也在处理项目的线程之间出现竞争条件，从而导致重复处理或更糟。

听起来你最好实际删除该项目并使用
责任链模式

编辑：回复：你的最后一个例子：如果你只有 1 个消费者，你永远不会删除队列中的对象 - 除非它同时更新 - 在这种情况下，你最好非常小心线程安全，并且可能不应该将项目放入无论如何都要排队。

Answer 7

看起来 BlockingQueue 本身没有您指定的功能。

不过，我可能会尝试重新思考一下问题：对于无法“正确处理”的对象，您会怎么做？如果您只是将它们留在队列中，则必须在某个时候将它们拉出来并处理它们。我建议要么弄清楚如何处理它们（通常，如果 queue.get() 给出任何类型的无效或错误值，您可能可以将其放在地板上）或选择不同的数据结构先进先出。

Answer 8

“最简单”的解决方案

在成功处理上一个元素之前，不要处理下一个元素。

public void run() {

Object lastSuccessfullyProcessedElement = null;

    while (!exit) {
        Object obj =  lastSuccessfullyProcessedElement == null ? queue.take() : lastSuccessfullyProcessedElement; // blocking
        
        boolean successful = process(obj);
        
        if(!successful) {
            lastSuccessfullyProcessedElement = obj;
        } else {
            lastSuccessfullyProcessedElement = null;
        }
    }
}

1. 调用peek()并检查是否值为 null 时 CPU 效率不高。

当以下程序的队列为空时，我发现系统上的 CPU 使用率达到 10%。

while (true) {
   Object o = queue.peek();
   if(o == null) continue;
   // omitted for the sake of brevity
}

2. 添加 sleep() 会增加速度。

3. 使用putLast将其添加回队列会扰乱顺序。而且，这是一个阻塞操作，需要锁。