一个生产者，多个消费者，有一些不寻常的转折
我有一个（Posix）服务器，它充当许多客户端到另一个上游服务器的代理。消息通常从上游服务器向下流动，然后与之匹配，并被推出到对该流量感兴趣的客户端的某个子集（维护来自上游服务器的FIFO顺序）。目前，这个代理服务器是使用事件循环的单线程（例如-select、epoll等），但现在我想将它变成多线程，这样代理就可以更充分地利用整个机器并实现更高的吞吐量。
我的高级设计是拥有一个由N个工作线程组成的池（其中N是机器上核数的一些小倍数），每个工作线程运行自己的事件循环。每个客户机连接将被分配给一个特定的工作线程，然后该工作线程将负责在客户机连接期间服务该客户机的所有I/O+超时需求。我还打算有一个单一的专用线程谁拉入消息从上游服务器。一旦消息被读入，它的内容可以被认为是不变的/不变的，直到它不再被需要和回收为止。工作人员从不更改消息内容--他们只是根据需要将其传递给客户。
我的第一个问题是：客户利益的匹配应该最好由生产者线程还是工人线程来完成？
我的第二个问题是：有什么好的方法来跟踪消息是否仍然被任何工作线程所需要？
一种简单的方法是为每条消息分配一个计数，即在第一次生成消息时仍需要多少工作线程来处理该消息。然后，当每个worker处理完消息后，它将以线程安全的方式减少计数，如果/当计数变为零时，我们将知道它可以被回收。
另一种方法是在消息传入时为它们分配64B个序列号，然后每个线程可以跟踪并记录它们以某种方式处理的最高序列号。然后，我们可以通过某种方式在所有工作线程中回收序号小于或等于最小处理序号的所有消息。
我知道无锁数据结构，但我不知道在我的情况下它们是否更可取，或者我是否应该对影响队列的操作使用简单的互斥体。此外，在共享队列方法中，我不完全确定工作线程应该如何跟踪它在处理队列中的“位置”。
任何见解都将不胜感激！谢了！
共有1个答案
根据您的问题描述，无论如何，需要为每个客户机为每个消息进行客户机兴趣匹配，因此无论它发生在哪种类型的线程中，匹配的工作都是相同的。这表明匹配应该在客户端线程中进行，以提高并发性。如果“生产者”线程确保在其他线程知道消息的可用性之前将消息刷新到主存（从技术上讲，“与其他线程同步内存”），那么同步开销不应该是一个主要问题，因为客户端线程都可以同时从主存读取信息，而无需彼此同步。客户端线程将不能修改消息，但它们应该不需要这样做。
消息回收可能通过跟踪每个线程的当前消息数来更好地完成，而不是通过使用消息特定计数器来完成，因为消息特定计数器会造成并发瓶颈。
我不认为你需要正式的排队机制。“生产者”线程可以简单地更新一个volatile变量，该变量包含已刷新到主存的最新消息的数量，客户端线程可以在空闲时检查该变量，如果没有可用的工作，则可以Hibernate。您可以在线程管理方面变得更加复杂，但额外的效率改进可能很小。
-
我有三根线。线程1(T1)是生成器，它生成数据。线程2和线程3(T2和T3)分别等待T1的数据在单独的循环中处理。我正在考虑在线程之间共享BlockingQueue,并通过调用“Take”让T2和T3等待。
-
我有一个生产者/消费者场景，我不希望一个生产者交付产品，多个消费者消费这些产品。然而，常见的情况是，交付的产品只被一个消费者消费，而其他消费者从未看到过这个特定的产品。我不想实现的是，一个产品被每个消费者消费一次，而没有任何形式的阻碍。 我的第一个想法是使用多个BlockingQueue，每个消费者使用一个，并使生产者将每个产品按顺序放入所有可用的BlockingQueues中。但是，如果其中一个
-
问题内容： 因此，我已经看到了许多在Go中实现一个消费者和许多生产者的方法-Go 并发中的经典fanIn函数。 我想要的是fanOut功能。它以一个通道作为参数，它从中读取一个值，并返回一个通道片，该通道将这个值的副本写入其中。 有没有正确/推荐的方法来实现这一目标？ 问题答案： 您几乎描述了执行此操作的最佳方法，但这是执行此操作的一小段代码示例。 去游乐场：https : //play.gola
-
我有一个使用ActiveMQ的消息队列。web请求用persistency=true将消息放入队列。现在，我有两个消费者，它们都作为单独的会话连接到这个队列。使用者1总是确认消息，但使用者2从不这样做。 JMS队列实现负载平衡器语义。一条消息将被一个使用者接收。如果在发送消息时没有可用的使用者，它将被保留，直到有可以处理消息的使用者可用为止。如果使用者接收到一条消息，但在关闭之前没有确认它，那么该
-
问题内容： 我有一个JMS客户端，它正在生成消息并通过JMS队列发送到其唯一的使用者。 我想要的是不止一个消费者收到这些消息。我想到的第一件事是将队列转换为主题，以便现有用户和新用户都可以订阅并将相同的消息传递给他们。 显然，这将涉及在生产者和消费者方面修改当前的客户代码。 我还要查看其他选项，例如创建第二个队列，这样就不必修改现有的使用者。我相信这种方法有很多优点，例如（如果我错了，请纠正我）在
-
由于消息需求的排序，我们有一个主题和一个分区。我们有两个消费者运行在不同的服务器上，具有相同的配置集，即groupId、consumerId和consumerGroup。即 1主题- 当我们部署消费者时，相同的代码会部署在两台服务器上。当消息到来时，我们会注意到两个消费者都在消费消息，而不是只有一个处理。让消费者在两台独立的服务器上运行的原因是，如果一台服务器崩溃，至少其他服务器可以继续处理消息。