指示某些值可用的 Java 信号/事件机制

Question

我有一个生成器类，它拥有一个线程，其中确定了要生成的“记录”数量，然后生成那么多记录(这些记录被放置在 BlockingQueue 中以供另一个线程检索)。

我希望其他线程知道将生成多少记录(用于合理的进度报告等)。

Future 似乎完全符合我的要求，但我是 Java 的新手，不确定实现它的惯用方式。

我的背景是 C++/Win32，所以我通常会使用 win32“事件”(由 CreateEvent(0, true, false, 0) 创建，带有 SetEvent 和 WaitForSingleObject 用于我的信号和等待实现)。我注意到 Java 有一个 CountDownLatch，但这在某种程度上感觉比我想要的更重(有点类似于当我真的想要一个 boolean 值时使用 int)，而且它似乎不直观(对我来说，无论如何)。

这是我使用 CountDownLatch 和 Future 的代码。我在这里稍微提炼了我的真实代码(删除了不相关的实现细节并忽略了所有错误处理)。

    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.Future;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.TimeUnit;

    public abstract class Generator {

        private CountDownLatch numRecordsSignal = new CountDownLatch(1);

        private int numRecords;

        private BlockingQueue<Record> queue = new LinkedBlockingQueue<Record>();

        public Generator() {
            new Thread(new Runnable() {

                @Override
                public void run() {
                    numRecords = calculateNumRecords();
                    numRecordsSignal.countDown();

                    for (Record r : generateRecords()) {
                        try {
                            queue.put(r);
                        } catch (InterruptedException e) {
                            // [ ... snip ... ]
                        }
                    }                
                }
            }).start();
        }

        public Future<Integer> numRecords() {
            return new Future<Integer>() {
                // Ignore cancel for now (It wouldn't make sense to cancel 
                // just this part of the Generator's work, anyway).
                public boolean cancel(boolean mayInterruptIfRunning) { 
                    return false; 
                }

                public Integer get() throws InterruptedException {
                    numRecordsSignal.await();
                    return numRecords;
                }

                public Integer get(long timeout, TimeUnit unit) 
                        throws InterruptedException {
                    numRecordsSignal.await(timeout, unit);
                    return numRecords;
                }

                public boolean isCancelled() {
                    return false;
                }

                public boolean isDone() {
                    // Since we can't cancel, just check the state of the 
                    // signal
                    return numRecordsSignal.getCount() == 0;
                }
            };
        }

        public Record nextRecord() throws InterruptedException {
            return queue.take();
        }

        /** --- Boring stuff below this line --- */
        public interface Record { }

        protected abstract int calculateNumRecords();

        protected abstract Iterable<Record> generateRecords();                        
    }

现在是我的实际问题:

1. 是否有比 CountDownLatch 更好的单发信号机制？
2. 我希望调用者能够等待或轮询结果，但不需要他们能够取消操作。 Future 是揭露这些东西的正确方式吗？
3. 这些东西看起来特别“非 Java”吗？我完全走错路了吗？

编辑:

澄清一下，我希望调用者能够执行以下操作:

    Generator gen = new Generator();
    Integer numRecords = gen.numRecords().get();  // This call might block waiting for the result
    numRecords = gen.numRecords().get(); // This call will never block, as the result is already available.

这只是我要实现的一个初始化缓慢的值。一旦满足“初始化”条件，它就应该锁存。该值一旦已知就不会重新评估。

Answer 1

旁注

You should not start a thread in a constructor - 可以想象，线程启动时 Generator 对象未完全创建，例如倒计时锁存器很可能为 null。您可以在构造函数中创建线程，但应该以单独的方法启动它。您的调用代码将变为:

Generator g = new Generator();
g.start();

您的问题

您正在自己重新实现一个 Future，在我看来这既没有必要也不可取。我会重新设计类(class)并制作 Generator实现 Callable<Integer>并通过执行者运行它。这为您提供了几样东西:

• 从生成器中删除线程逻辑，这使您能够在调用堆栈的更高级别更有效地管理线程
• 整数在您的调用代码中通过 future 返回，您依赖 JDK 来处理实现
• 我假设可以先填充队列然后返回整数
• 您可以调用future.get()任意多次 - 它只会在第一次调用时阻塞。

public static void main(String[] args) {
    ExecutorService executor = Executors.newFixedThreadPool(1);
    Future<Integer> future = executor.submit(new GeneratorImpl()); //a concrete implementation of Generator
    int numRecords = 0;
    try {
        numRecords = future.get(); //you can use a get with timeout here
    } catch (ExecutionException e) {
        //an exception happened in Generator#call()
    } catch (InterruptedException e) {
        //handle it
    }

    //don't forget to call executor.shutdown() when you don't need it any longer
}

public abstract class Generator implements Callable<Integer> {

    private BlockingQueue<Record> queue = new LinkedBlockingQueue<Record>();

    @Override
    public Integer call() {
        int numRecords = calculateNumRecords();
        for (Record r : generateRecords()) {
            try {
                queue.put(r);
            } catch (InterruptedException e) {
                // [ ... snip ... ]
            }
        }
        return numRecords;
    }

    public Record nextRecord() throws InterruptedException {
        return queue.take();
    }

    /**
     * --- Boring stuff below this line ---
     */
    public interface Record {
    }

    protected abstract int calculateNumRecords();

    protected abstract Iterable<Record> generateRecords();
}

---

编辑

如果您需要尽快返回 numRecods，您可以在单独的线程中填充您的队列:

    public Integer call() {
        int numRecords = calculateNumRecords();
        new Thread(new Runnable() {
            @Override
            public void run() {
                for (Record r : generateRecords()) {
                    try {
                        queue.put(r);
                    } catch (InterruptedException e) {
                        // [ ... snip ... ]
                    }
                }
            }
        }).start(); //returns immediately
        return numRecords;
    }