如何将线程分配给一种数据？答案

【问题标题】：How to Assign a Thread to a Type of Data?如何将线程分配给一种数据？
【发布时间】：2013-10-09 14:57:50
【问题描述】：

我有两种类型的顺序数据流。而且，每种类型的数据都有一个订阅（监听）的观察者。

Observer observer1 = new Observer() {
    @Override
    public void next(Data data) {
    //Do something
    }

}


Observer observer2 = new Observer() {
    @Override
    public void next(Data data) {
    //Do something
    }

}

feed.subscribe(observer1, "Type1");
feed.subscribe(observer2, "Type2");

当收到Type1的一条数据时，会调用observer1的next()方法，Type2和observer2也是一样。

由于数据是顺序的，当我们有type1和type2的数据时，第二个应该等待第一个的进程（将被阻塞）。

现在我想用多线程来解决这个问题。但是，我不想为收到的每条数据创建一个线程。我希望观察者并行运行。

例如，如果我有这个序列：

type1, type1, type1, type2, ...

第二个和第三个数据应该等待第一个处理（因为它们是type1），但第四个应该由observer2收集并处理。

请注意，我使用的 API 只允许我订阅，它将处理 next() 方法。

我想到的解决方案是对线程和观察者进行映射。因此，在每次调用next() 方法时，我可以调用分配给该类型的线程来恢复，而不是创建新的Thread。或者，如果有一个线程为observer1 运行，我希望type1 的所有数据都留在队列中等待该线程处理它们。

在Java中解决这个问题的任何设计模式或数据结构？

【问题讨论】：

当然你应该使用 Executors.newCachedThreadPool() 而不是分叉你自己的线程，但这并不能说明排序是如何发生的。
两者 Observers 期望相同的 Type 即 Data。
@JarrodRoberson，是的，API 将根据我订阅的类型选择正确的观察者。
致downvoter(s)：原因是哪一个：没有研究工作、没有用或不清楚。请告知更多问题。谢谢。
@Afshin Moazami 没关系，你的问题是合理的。在没有解释的情况下投反对票确实是不礼貌的表现。

标签： java multithreading observer-pattern producer-consumer

【解决方案1】：

这听起来很适合producer/consumer pattern。会有两个消费者，每个数据类型一个。还有一个生产者线程，它是数据馈送的订阅者。对于接收到的每个数据元素，它会将一个工作单元放在队列中以供适当的消费者使用。上面的 Wikipedia 文章有一个示例 Java 实现。网络上还提供了许多其他资源。

【讨论】：

【解决方案2】：

这是通过队列解决的典型问题。有一个消费者线程、两个队列（通常是 N 个队列，其中 N 是您的数据类型）和 2 个（通常也是 N 个）工作线程。消费者线程读取流并将“消息”放入正确的队列中。每个工作线程轮询其对应的队列，处理下一个可用项并重复。

参考：查看java.util.Queue for Java >= 1.5 的实现。

【讨论】：