包含 java.util.concurrent.BlockingQueue 的谓词 (/find/search)？答案

【问题标题】：Contains with a predicate (/find/search) for a java.util.concurrent.BlockingQueue?包含 java.util.concurrent.BlockingQueue 的谓词 (/find/search)？
【发布时间】：2017-03-29 18:34:31
【问题描述】：

java.util.concurrent.BlockingQueue 有一个方便的contains 方法：

/**
 * Returns {@code true} if this queue contains the specified element.
 * More formally, returns {@code true} if and only if this queue contains
 * at least one element {@code e} such that {@code o.equals(e)}.
 *
 * @param o object to be checked for containment in this queue
 * @return {@code true} if this queue contains the specified element
 * @throws ClassCastException if the class of the specified element
 *         is incompatible with this queue
 *         (<a href="../Collection.html#optional-restrictions">optional</a>)
 * @throws NullPointerException if the specified element is null
 *         (<a href="../Collection.html#optional-restrictions">optional</a>)
 */
public boolean contains(Object o);

我的需要更具体一点：应用predicate / 搜索条件。有什么办法可以做到这一点

  Using a ArrayBlockQueue
  Invoking toArray

这会起作用..但是如果队列很大怎么办？这可能会导致内存分配问题。

【问题讨论】：

这对于多个队列来说似乎是一个很好的例子，而不是试图将一个队列硬塞到一些奇怪的、非队列式的行为中。
@pvg 搜索基于在消息中查找特定 ID。基于固定谓词的“多个”队列并非如此。采用这种方法时，我们需要为每条消息创建一个队列。
如果消息 ID 是唯一的，您只需覆盖 .equals。或者设置某种优先级。如果您必须识别特定消息，那么使用强加命令的队列有什么意义？您应该描述您要解决的问题，而不是“如何使队列表现得像非队列”。
'你所要做的就是覆盖 .equals。'容器是一个匿名元组：它需要将队列的元素更改为不同的类。也许这是一个合理的方法 - 但它与问题不同。回复：使队列像非队列一样。我在过去十年中已经构建了自定义排队系统。这是一个队列，要求进行搜索并不能让它不是这样。
要求它进行搜索是要求无序访问。当然，您可以将其添加到队列中。或者您可以使用另一个数据结构来跟踪您想要乱序访问的任何内容。但是你不会在队列中找到很多对这类东西的丰富支持，因为一般来说，这不是队列的主要工作。如您所见，反映在 JDK 提供的队列中。

标签： java queue blockingqueue

【解决方案1】：

我不知道你是否应该这样做，但似乎你可以这样做：

public static class Matcher<T> {

    Predicate<T> predicate;
    Class<T> clazz;

    public Matcher(Predicate<T> t, Class<T> clazz) {
        this.predicate = t;
        this.clazz = clazz;
    }

    @SuppressWarnings("unchecked")
    public boolean equals(Object o) {
        if (o != null && this.clazz.isInstance(o)) {
            System.out.println("Checking item " + o + " against predicate " + this.predicate);
            return this.predicate.test((T) o);
        }
        return false;
    }
}

public static void main(String args[]) {

    ArrayBlockingQueue<Integer> oddsOnly = new ArrayBlockingQueue<>(10);
    oddsOnly.addAll(Arrays.asList(1, 3, 5, 7, 9, 11, 13));

    ArrayBlockingQueue<Integer> oddsAndEves = new ArrayBlockingQueue<>(10);
    oddsAndEves.addAll(Arrays.asList(1, 2, 3, 4, 5));

    Predicate<Integer> evenPredicate = (i) -> i % 2 == 0;
    System.out.println(oddsOnly.contains(new Matcher<>(evenPredicate, Integer.class))); // prints false

    System.out.println(oddsAndEves.contains(new Matcher<>(evenPredicate, Integer.class))); // prints true
}

输出：

Checking item 1 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
Checking item 3 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
Checking item 5 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
Checking item 7 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
Checking item 9 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
Checking item 11 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
Checking item 13 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
false
Checking item 1 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
Checking item 2 against predicate QueueEquals$Matcher$$Lambda$1/531885035@816f27d
true

它确实让我想洗个澡....另外，正如pvg 正确指出的那样 - 这不能保证有效。 BlockingQueue 实现可能会在调用equals() 之前调用hashCode()，然后我们就完蛋了。

一个更明智的选择是只遍历阻塞队列，并对每个元素运行谓词。 BlockingQueue 接口没有iterator，但很多阻塞队列实现，例如ArrayBlockingQueue 和LinkedBlockingQueue 都有。

【讨论】：

这太可怕了！但也不能“保证”（在迂腐的意义上）工作，因为不能保证 Collection 实现调用 equals。 “本规范不应被解释为暗示使用非空参数 o 调用 Collection.contains 将导致对任何元素 e 调用 o.equals(e)”
我也有同感。这很可怕，但它适用于在调用 equals 之前不调用 hashCode 的实现。无论如何，我在答案的底部添加了一个更明智的解决方案。

【解决方案2】：

您可以通过Queue 流式传输（如果您愿意，可以并行）并使用您的谓词过滤掉您的元素。 BlockingQueue 是 Queue，它是 Collection，因此定义了 stream() 和 parallelStream()。以下内容：

public class SampleApplication {

    private static final int ELEMENT_COUNT = 100000;

    public static void main(String[] args) {
        Queue<Element> blockingQueue = new ArrayBlockingQueue<>(ELEMENT_COUNT);
        List<Element> elements = Stream.generate(SampleApplication::createElement).limit(ELEMENT_COUNT).collect(Collectors.toList());
        blockingQueue.addAll(elements);

        blockingQueue.stream().filter(getFilterPredicate(5)).findAny().ifPresent(System.out::println);
        blockingQueue.stream().filter(getFilterPredicate(105)).findAny().ifPresent(System.out::println);
        blockingQueue.stream().filter(getFilterPredicate(-1)).findAny().ifPresent(System.out::println);
    }

    private static Predicate<Element> getFilterPredicate(int value) {
        return e -> e.getId() == value;
    }

    private static int i = 0;
    private static Element createElement() {
        return new Element(++i);
    }

    static class Element {
        private int id;
        private String message;

        public Element(int id) {
            this.id = id;
            this.message = "Default msg";
        }

        //Getters, Setters & toString omitted
    }
}

具有以下输出找到的值：

Element{id=5, message=Default msg}
Element{id=105, message=Default msg}

【讨论】：