生产者-消费者与谓词

Question

我正在寻找一个支持在谓词上阻塞read()s 的Java 集合。我写了一个简单的版本，但它似乎已经被发明了？

例如：

interface PredicateConsumerCollection<T> {

  public void put(T t);

  @Nullable
  public T get(Predicate<T> p, long millis) throws InterruptedException;
}

put() 将其参数传递给具有匹配谓词的等待消费者，或将其存储在商店中。如果合适的T 已经在商店中，则get() 立即返回，或者阻塞直到合适的值被放置（），或者超时。消费者竞争，但公平对我来说并不重要。

有人知道这样的收藏吗？

Answer 1

没有直接的类可以解决您的问题，但 ConcurrentHashMap 和 BlockingQueue 的组合可能是一个解决方案。

哈希映射定义为：

final ConcurrentHashMap<Predicate, LinkedBlockingQueue<Result>> lookup;

put 需要确保为每个 Predicate 添加一个队列到映射中，这可以使用putIfAbsent 线程安全地完成。

如果你有一组固定的谓词，你可以简单地预先填充列表，然后消费者可以简单地调用lookup.get(Predicate).take()

如果 Predicate 的数量未知/太多，您需要为消费者编写等待/通知实现，以防 Predicate 尚未在您自己的列表中。

Answer 2

我还需要一些非常相似的东西来测试在某个超时时间内是否收到了某个 JMS 异步消息。事实证明，通过使用Oracle tutorials 中解释的基本等待/通知，您的问题相对容易实现。这个想法是使 put 和 query 方法同步，并让 query 方法等待。 put 方法调用 notifyAll 来唤醒查询方法中的所有等待线程。然后查询方法必须检查谓词是否匹配。最棘手的事情是由于在谓词不匹配时唤醒以及由于可能的“虚假唤醒”而正确超时。我发现this stackoverflow post 提供了答案。

这是我想出的实现：

import java.util.ArrayList;
import java.util.List;

// import net.jcip.annotations.GuardedBy;

import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;

public class PredicateConsumerCollectionImpl<T> implements
        PredicateConsumerCollection<T> {

    // @GuardedBy("this")
    private List<T> elements = new ArrayList<>();

    @Override
    public synchronized void put(T t) {
        elements.add(t);
        notifyAll();
    }

        @Override
public synchronized T query(Predicate<T> p, long millis)
        throws InterruptedException {
    T match = null;
    long nanosOfOneMilli = 1000000L;
    long endTime = System.nanoTime() + millis * nanosOfOneMilli;
    while ((match = Iterables.find(elements, p, null)) == null) {
        long sleepTime = endTime - System.nanoTime();
        if (sleepTime <= 0) {
            return null;
        }
        wait(sleepTime / nanosOfOneMilli,
                (int) (sleepTime % nanosOfOneMilli));
    }
    return match;
}

    synchronized boolean contains(T t) {
        return elements.contains(t);
    }
}

这是一个 JUnit 测试，证明代码按预期工作：

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;

import org.junit.Before;
import org.junit.Test;

import com.google.common.base.Predicate;

/**
 * Unit test for the {@link PredicateConsumerCollection} implementation.
 * 
 * <p>
 * The tests act as consumers waiting for the test Producer to put a certain
 * String.
 */
public class PredicateConsumerCollectionTest {

    private static class Producer implements Runnable {

        private PredicateConsumerCollection<String> collection;

        public Producer(PredicateConsumerCollection<String> collection) {
            this.collection = collection;
            collection.put("Initial");
        }

        @Override
        public void run() {
            try {
                int millis = 50;
                collection.put("Hello");
                Thread.sleep(millis);
                collection.put("I");
                Thread.sleep(millis);
                collection.put("am");
                Thread.sleep(millis);
                collection.put("done");
                Thread.sleep(millis);
                collection.put("so");
                Thread.sleep(millis);
                collection.put("goodbye!");
            } catch (InterruptedException e) {
                e.printStackTrace();
                fail("Unexpected InterruptedException");
            }
        }

    }

    private PredicateConsumerCollectionImpl<String> collection;
    private Producer producer;

    @Before
    public void setup() {
        collection = new PredicateConsumerCollectionImpl<>();
        producer = new Producer(collection);
    }

    @Test(timeout = 2000)
    public void wait_for_done() throws InterruptedException {
        assertTrue(collection.contains("Initial"));
        assertFalse(collection.contains("Hello"));

        Thread producerThread = new Thread(producer);
        producerThread.start();

        String result = collection.query(new Predicate<String>() {
            @Override
            public boolean apply(String s) {
                return "done".equals(s);
            }
        }, 1000);
        assertEquals("done", result);
        assertTrue(collection.contains("Hello"));
        assertTrue(collection.contains("done"));

        assertTrue(producerThread.isAlive());
        assertFalse(collection.contains("goodbye!"));

        producerThread.join();

        assertTrue(collection.contains("goodbye!"));
    }

    @Test(timeout = 2000)
    public void wait_for_done_immediately_happens() throws InterruptedException {
        Thread producerThread = new Thread(producer);
        producerThread.start();

        String result = collection.query(new Predicate<String>() {
            @Override
            public boolean apply(String s) {
                return "Initial".equals(s);
            }
        }, 1000);
        assertEquals("Initial", result);
        assertFalse(collection.contains("I"));

        producerThread.join();

        assertTrue(collection.contains("goodbye!"));
    }

    @Test(timeout = 2000)
    public void wait_for_done_never_happens() throws InterruptedException {
        Thread producerThread = new Thread(producer);
        producerThread.start();

        assertTrue(producerThread.isAlive());

        String result = collection.query(new Predicate<String>() {
            @Override
            public boolean apply(String s) {
                return "DONE".equals(s);
            }
        }, 1000);

        assertEquals(null, result);
        assertFalse(producerThread.isAlive());
        assertTrue(collection.contains("goodbye!"));
    }

}