使用BlockingCollections在队列中运行多个线程

Question

我的程序有3个功能.每个函数都会获取一个项目列表并填充某些信息.例如

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

函数F1采用List并填充upc

功能F2取List(F1的输出)并填写价格.

功能F3取List(输出F2)并填充competitorName

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

如果您有连续的项目进入 F1，这是Apache Storm的典型用例。您可以在几分钟内在 Storm 中实现这一点，并且您将拥有快速且完美的并行系统。你的 F1、F2 和 F3 将变成螺栓，你的物品生产者将变成喷水。

既然您问如何使用 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的基本架构