task-parallel-library - 使用 BlockingCollections 在队列中运行多个线程

Question

我的程序有 3 个功能。每个函数都接受一个项目列表并填充某些信息。
例如

class Item {
 String sku,upc,competitorName;
 double price;
}

功能 F1 获取一个 List 并填充 upc

功能 F2 获取列表(F1 的输出)并填充 价格 .

功能 F3 获取列表(F2 的输出)并填充 参赛者姓名

F1 一次可以处理5个项目，
F2 一次可以处理20个项目，
F3 还有20。

现在我正在串行运行 F1 -> F2 -> F3，因为 F2 需要来自 F1 的信息(UPC 代码)。 F3 需要 F2 的价格。

我想通过连续运行 F1 而不是等待 F2 和 F3 完成来使这个过程有效。 F1 执行并输出到队列中，然后 F2 一次取 20 个项目并处理它们。然后跟随F3。

我如何通过使用 BlockingCollection 和 Queue 来实现这一点？

Answer 1

这是 的典型用例 Apache Storm 如果您有连续的元素进入 F1。您可以在几分钟内在 Storm 中实现这一点，并且您将拥有快速且完美的并行系统。您的 F1、F2 和 F3 将成为 bolt ，而您的元素生产商将成为喷口。

既然你问如何使用 BlockingCollections 来做到这一点，这里是一个实现。您总共需要 3 个线程。

ItemsProducer:它一次生产 5 个项目并将其提供给 F1。

F2ExecutorThread:它一次消耗 20 件元素并将其喂给 F2。

F3ExecutorThread:它一次消耗 20 个项目并将其提供给 F3。

您还有 2 个阻塞队列，一个用于从 F1->F2 传输数据，另一个用于从 F2->F3 传输数据。如果需要，您还可以有一个队列以类似的方式将数据提供给 F1。这取决于您如何获得元素。我使用 Thread.sleep 来模拟执行函数所需的时间。

每个函数将继续在其分配的队列中查找项目，而不管其他函数在做什么，并等待队列中有项目。一旦他们处理了这个项目，他们就会把它放在另一个队列中以供另一个功能使用。如果另一个队列已满，他们将等到另一个队列有空间。

由于您的所有函数都在不同的线程中运行，因此 F1 不会等待 F2 或 F3 完成。如果您的 F2 和 F3 明显快于 F1，您可以将更多线程分配给 F1 并继续推送到相同的 f2Queue。

public class App {

    final BlockingQueue<Item> f2Queue = new ArrayBlockingQueue<>(100);
    final BlockingQueue<Item> f3Queue = new ArrayBlockingQueue<>(100);

    public static void main(String[] args) throws InterruptedException {
        App app = new App();
        app.start();
    }

    public void start() throws InterruptedException {
        Thread t1 = new ItemsProducer(f2Queue);
        Thread t2 = new F2ExecutorThread(f2Queue, f3Queue);
        Thread t3 = new F3ExecutorThread(f3Queue);

        t1.start();
        t2.start();
        t3.start();

        t1.join();
        t2.join();
        t3.join();
    }
}

/**
 * Thread producing 5 items at a time and feeding it to f1()
 */
class ItemsProducer extends Thread {
    private BlockingQueue<Item> f2Queue;

    private static final int F1_BATCH_SIZE = 5;

    public ItemsProducer(BlockingQueue<Item> f2Queue) {
        this.f2Queue = f2Queue;
    }

    public void run() {
        Random random = new Random();
        while (true) {
            try {
                List<Item> items = new ArrayList<>();
                for (int i = 0; i < F1_BATCH_SIZE; i++) {
                    Item item = new Item(String.valueOf(random.nextInt(100)));
                    Thread.sleep(20);
                    items.add(item);
                    System.out.println("Item produced: " + item);
                }

                // Feed items to f1
                f1(items);

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    void f1(List<Item> items) throws InterruptedException {
        Random random = new Random();
        for (Item item : items) {
            Thread.sleep(100);
            item.upc = String.valueOf(random.nextInt(100));
            f2Queue.put(item);
        }
    }
}

/**
 * Thread consuming items produced by f1(). It takes 20 items at a time, but if they are not
 * available it waits and starts processesing as soon as one gets available
 */
class F2ExecutorThread extends Thread {
    static final int F2_BATCH_SIZE = 20;
    private BlockingQueue<Item> f2Queue;
    private BlockingQueue<Item> f3Queue;

    public F2ExecutorThread(BlockingQueue<Item> f2Queue, BlockingQueue<Item> f3Queue) {
        this.f2Queue = f2Queue;
        this.f3Queue = f3Queue;
    }

    public void run() {
        try {
            List<Item> items = new ArrayList<>();
            while (true) {
                items.clear();
                if (f2Queue.drainTo(items, F2_BATCH_SIZE) == 0) {
                    items.add(f2Queue.take());
                }
                f2(items);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    void f2(List<Item> items) throws InterruptedException {
        Random random = new Random();
        for (Item item : items) {
            Thread.sleep(100);
            item.price = random.nextInt(100);
            f3Queue.put(item);
        }
    }
}

/**
 * Thread consuming items produced by f2(). It takes 20 items at a time, but if they are not
 * available it waits and starts processesing as soon as one gets available.
 */
class F3ExecutorThread extends Thread {
    static final int F3_BATCH_SIZE = 20;
    private BlockingQueue<Item> f3Queue;

    public F3ExecutorThread(BlockingQueue<Item> f3Queue) {
        this.f3Queue = f3Queue;
    }

    public void run() {
        try {
            List<Item> items = new ArrayList<>();
            while (true) {
                items.clear();
                if (f3Queue.drainTo(items, F3_BATCH_SIZE) == 0) {
                    items.add(f3Queue.take());
                }
                f3(items);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void f3(List<Item> items) throws InterruptedException {
        Random random = new Random();

        for (Item item : items) {
            Thread.sleep(100);
            item.competitorName = String.valueOf(random.nextInt(100));
            System.out.println("Item done: " + item);
        }
    }
}

class Item {
    String sku, upc, competitorName;
    double price;

    public Item(String sku) {
        this.sku = sku;
    }

    public String toString() {
        return "sku: " + sku + " upc: " + upc + " price: " + price + " compName: " + competitorName;
    }
}

我想你也可以在 .Net 中遵循完全相同的方法。为了更好地理解，我建议您阅读 http://storm.apache.org/releases/current/Tutorial.html 的基本架构