有效阻塞查询的建议

这是一些源代码。它基本上与CB160所说的相同,但是拥有源代码可能有助于清除您可能仍然存在的任何问题。尤其是FunkyCollection上的方法必须同步。

正如Meriton指出的,get方法在每次添加新对象时对每个阻塞的get执行O(N)扫描。它还执行O(N)操作来删除对象。这可以通过使用类似于链表的数据结构来改进,在链表中可以检查最后一个项目的迭代器。我没有为这个优化提供源代码,但是如果您需要额外的性能,实现它应该不会太困难。

import java.util.*;

public class BlockingQueries
{
    public class Pair
    {
        public final String first;
        public final String second;
        public Pair(String first, String second)
        {
            this.first = first;
            this.second = second;
        }
    }

    public class FunkyCollection
    {
        final ArrayList<Pair> pairs = new ArrayList<Pair>();

        public synchronized void add( Pair p )
        {
            pairs.add(p);
            notifyAll();
        }

        public synchronized Pair get( Pair p ) throws InterruptedException
        {
            while (true)
            {
                for (Iterator<Pair> i = pairs.iterator(); i.hasNext(); )
                {
                    Pair pair = i.next();
                    boolean firstOk = p.first == null || p.first.equals(pair.first);
                    boolean secondOk = p.second == null || p.second.equals(pair.second);
                    if (firstOk && secondOk)
                    {
                        i.remove();
                        return pair;                
                    }
                }
                wait();
            }
        }   
    }

    class Producer implements Runnable
    {
        private FunkyCollection funkyCollection;

        public Producer(FunkyCollection funkyCollection)
        {
            this.funkyCollection = funkyCollection;
        }

        public void run()
        {
            try
            {
                for (int i = 0; i < 10; ++i)
                {
                    System.out.println("Adding item " + i);
                    funkyCollection.add(new Pair("foo" + i, "bar" + i));
                    Thread.sleep(1000);
                }
            }
            catch (InterruptedException e)
            {
                Thread.currentThread().interrupt();
            }
        }
    }

    public void go() throws InterruptedException
    {
        FunkyCollection funkyCollection = new FunkyCollection();
        new Thread(new Producer(funkyCollection)).start();
        System.out.println("Fetching bar5.");
        funkyCollection.get(new Pair(null, "bar5"));
        System.out.println("Fetching foo2.");
        funkyCollection.get(new Pair("foo2", null));
        System.out.println("Fetching foo8, bar8");
        funkyCollection.get(new Pair("foo8", "bar8"));
        System.out.println("Finished.");
    }

    public static void main(String[] args) throws InterruptedException
    {
        new BlockingQueries().go();
    }
}

输出:

Fetching bar5.
Adding item 0
Adding item 1
Adding item 2
Adding item 3
Adding item 4
Adding item 5
Fetching foo2.
Fetching foo8, bar8
Adding item 6
Adding item 7
Adding item 8
Finished.
Adding item 9

请注意,我将所有内容都放在一个源文件中,以便于运行。

我知道没有任何现有的容器可以提供这种行为。您面临的一个问题是,没有现有条目与查询匹配。在这种情况下,您必须等待新条目到达,而这些新条目应该到达序列的尾部。考虑到您正在阻塞,您不需要检查最新添加之前的所有条目,因为您已经检查了它们并确定它们不匹配。因此,您需要某种方法来记录您当前的位置,并且能够在新条目到达时从那里向前搜索。

这个等待是一个工作 Condition . 如建议 cb160 's answer ,您应该分配一个 条件 集合中的实例,并通过 Condition#await() .你也应该让你的同伴承受你的压力。 get() 允许定时等待的方法:

public Pair get(Pair p) throws InterruptedException;
public Pair get(Pair p, long time, TimeUnit unit) throws InterruptedException;

每次呼叫 add() 呼吁 Condition#signalAll() 取消阻止等待不满意的线程 获取() 查询,允许他们扫描最近添加的内容。

您没有提到如何或是否从该容器中移除项目。如果容器只增长,这就简化了线程扫描其内容的方式,而不必担心来自其他线程的争用会改变容器。每个线程都可以自信地开始查询 最低限度 可供检查的条目数。但是,如果您允许删除项目,那么还需要面对更多的挑战。

在FunkyCollection Add方法中,可以在每次添加元素时对集合本身调用NotifyAll。

在get方法中,如果底层容器(任何合适的conatiner都可以)不包含所需的值,请等待funkycollection。通知等待后,检查基础容器是否包含所需的结果。如果是,则返回该值,否则,请再次等待。

似乎您正在寻找元组空间的实现。这个 Wikipedia article about them 列出Java的几种实现方式,也许您可以使用其中之一。否则,您可能会发现一个要模仿的开源实现,或者相关的研究论文。