java - 保留已完成工作项顺序的多线程执行

Question

我有一个工作单元流，我们称它们为按顺序处理的“工作项”(暂时)。我想通过多线程处理来加快处理速度。

约束:这些工作项按特定顺序出现，在处理过程中顺序不相关 - 但一旦处理完成，必须恢复顺序。

像这样:

   |.|
   |.|
   |4|
   |3|
   |2|    <- incoming queue
   |1|
  / | \
 2  1  3  <- worker threads
  \ | /
   |3|
   |2|    <- outgoing queue
   |1|

我想用Java来解决这个问题，最好不用Executor Services、Futures等，而是用wait()、notify()等基本的并发方法

原因是:我的工作项非常小且粒度细，它们每个大约在 0.2 毫秒内完成处理。所以我担心使用 java.util.concurrent.* 中的内容可能会导致大量开销并降低我的代码速度。

到目前为止我找到的例子都在处理过程中保留了顺序(这对我来说无关紧要)并且不关心处理后的顺序(这对我来说很重要)。

Answer 1

这就是我在之前的项目中解决您的问题的方法(但是使用 java.util.concurrent):

(1) WorkItem 类进行实际工作/处理:

public class WorkItem implements Callable<WorkItem> {
    Object content;
    public WorkItem(Object content) {
        super();
        this.content = content;
    }

    public WorkItem call() throws Exception {
        // getContent() + do your processing
        return this;
    }
}

(2) 该类将Work Items放入队列并启动处理:

public class Producer {
    ...
    public Producer() {
        super();
        workerQueue = new ArrayBlockingQueue<Future<WorkItem>>(THREADS_TO_USE);
        completionService = new ExecutorCompletionService<WorkItem>(Executors.newFixedThreadPool(THREADS_TO_USE));
        workerThread = new Thread(new Worker(workerQueue));
        workerThread.start();
    }

    public void send(Object o) throws Exception {
        WorkItem workItem = new WorkItem(o);
        Future<WorkItem> future = completionService.submit(workItem);
        workerQueue.put(future);
    }
}

(3) 一旦处理完成，工作项就在这里出队:

public class Worker implements Runnable {
    private ArrayBlockingQueue<Future<WorkItem>> workerQueue = null;

    public Worker(ArrayBlockingQueue<Future<WorkItem>> workerQueue) {
        super();
        this.workerQueue = workerQueue;
    }

    public void run() {
        while (true) {
            Future<WorkItem> fwi = workerQueue.take(); // deqeueue it
            fwi.get(); // wait for it till it has finished processing
        }
    }
}

(4) 这是您在代码中使用这些东西并提交新工作的方式:

public class MainApp {
    public static void main(String[] args) throws Exception {
        Producer p = new Producer();
        for (int i = 0; i < 10000; i++)
            p.send(i);
    }
}