问题标签 [producer-consumer]

我有几个问题。
1.在内部网络进程通信的wcf自助服务中，一个服务可以在最短的时间内处理多少个调用？
2.是否有人为生产者/消费者的通用包装器提供了开源代码，具有负载平衡甚至是普通示例？
盖伊

我正在尝试使用 BlockingQueue 构建一个通用的生产者/消费者。
我希望它尽可能多线程或并行，但不占用所有计算机资源。
假设我们有一个生产者，在消费数据时将消费者作为线程还是一个消费者与任务更好？

或者

有人可以确认我在这方面是对还是错。在我看来，“选择器”操作是在 MessageConsumer 实现中完成的。（即所有消息仍从 Message Broker 分派到 MessageConsumer，然后针对这些消息执行“选择器”操作）。当我们有一堆我们不感兴趣的消息（即不匹配我们的选择器）时，就会出现问题，这些消息最终会由于预取或缓存限制而填满 MessageConsumer 的内部队列。因此，我们将无法接收任何新消息，尤其是我们对选择器感兴趣的消息。

那么，有没有办法配置 AMQ 在 MessageBroker 端执行选择器操作呢？我应该开始查看“拦截器”并创建自己的 BrokerPlugin 吗？有关如何解决此问题的任何建议？

我真的很感激任何答案。

谢谢，Soonthorn A.

现在，我有一个队列，有多个生产者和一个消费者。

消费者线程操作很慢。另外，消费者通过 peek 操作从队列中取出元素，直到消费操作完成，才能将元素从队列中移除。这是因为作为辅助操作的生产者线程还拍摄了当时未完全处理的所有元素的快照。

现在，我想更改我的代码以支持多个消费者。所以，假设我有三个线程，一个线程将获取第一个元素，可以通过 peek 操作读取。第二个消费者线程可以获取第二个元素，但我无法检索它，因为队列不支持检索第二个元素。

因此，使用标准 ConcurrentLinkedQueue （我现在正在使用）的选项已经取消。

我正在考虑使用优先级队列，但是我必须将每个元素与一个标志相关联，该标志告诉我该元素是否已被某个线程使用。

哪种数据结构最适合这个问题？

我有一个系统，可以从列表中读取名称，调用外部服务器进行真/假状态检查，并对具有真实状态的人采取行动。对外部服务器的调用需要一些时间，因此在一个线程中运行它并不是很有效。

我目前正在尝试将其实现为生产者/消费者系统，其中许多消费者线程从列表中读取名称，调用外部服务器，将有效名称放入阻塞队列中，并让单个消费者从队列中挑选项目并采取行动. 遗憾的是，系统有时会运行完成，有时会无限期挂起。测试代码如下

请告诉我哪里出错了。

我有一个系统，通过 HTTP 调用我的 PHP 代码（生产者）获取请求。此代码将请求参数添加到 MYSQL(queue) 中的表中。然后由 java 程序（消费者）获取和处理。在我的第一个实现中，生产者和消费者都在 PHP（使用 MYSQL 队列）中。然后随着负载的增加，这被证明是低效的，所以我制作了消费者 java。现在我认为从我的 java 应用程序中轮询队列的 MYSQL 表效率低下（MYSQL 进程的 cpu 使用率很高）。有没有更好的方法来实现这个队列（在 PHP 代码和 Java 应用程序之间共享内存或其他东西）？

我有典型的生产者-消费者应用程序。我有 BlockingCollection 来提供生产消费机制。我需要的是添加从另一个线程暂停消费者线程执行的能力。我已经阅读了不同的 .net 同步原语，但我无法决定如何以最简单的方式实现它。

PS：我使用.net 4.0。

我正在尝试使用一组资源来实现生产者/消费者模式，因此每个线程都有一个与之关联的资源。例如，我可能有一个任务队列，其中每个任务都需要StreamWriter写入其结果。每个任务还​​必须有参数传递给它。

我从 Joseph Albahari 的实现开始（见下文我的修改版本）。

我用资源在哪里的Action队列替换了队列，并将与线程关联的资源传递给. 但是，这给我留下了如何将参数传递给. 显然，必须用委托替换，但这留下了在任务入队时（从类外部）如何传递参数的问题。关于如何做到这一点的任何想法？Action<T>TActionActionActionProducerConsumerQueue

我不确定在以下情况下会发生什么，如果有人解释它会很有帮助。

生产者（P）发送消息 M1、M2、M3、M4、M5、M6、M7；假设它是从时间 T1 发送到 T7。

假设 ActiveMQ 被用作 JMS Provider (AMQ)

消息消费者 (L) 将是绑定到主题的侦听器。主题名称：jmsTopic

方案 1：

AMQ 正在运行，L 没有连接到 AMQ 中的 jmsTopic。P 将 M1 到 M7 发送到 jmsTopic。如果 L 在时间 T8 连接到 jmsTopic，它大于 T7，它会接收消息 M1 到 M7 还是只接收 T8 时间之后发送到主题的消息。

场景2：AMQ正在运行，L连接并监听jmsTopic，P发送M1到M4。L 接收 M1 到 M4。同时L处理M1到M4，P发送M5到M7，但是L在处理M4的过程中崩溃了。如果 L 再次将自己连接到 jmsTopic ，它会收到 M5 到 M7 还是只有在 L 连接到 jmsTopic 之后发送的消息才会被 L 接收。

场景3：AMQ正在运行，L连接并监听jmsTopic，P发送M1到M7。但是 AMQ 崩溃了；是否知道 AMQ 状态并在 AMQ 启动并运行后重新连接。

我正在尝试通过 HTTP 推送通知消费者编写一个 XML，它接受推送通知并解析 XML 并将数据保存在数据库中。我看过 Nokogiri gem，它是一个简洁的 XML 解析器。令我失望的是，我计划仅在我的应用程序上的用户订阅接收他们的活动提要时才接受通知。

所以，我期待这种格式的通知 <?xml version="1.0" encoding="UTF-8"?> <Activities xmlns=""> <RecordSet> <Record> <ACTIVITY_DATE>2011-05-10</ACTIVITY_DATE> <ACTIVITY_TYPE>FOO</ACTIVITY_TYPE> <ACCOUNT_ID>0123456789</ACCOUNT_ID> <SENDER_FIRST_NAME>John</SENDER_FIRST_NAME> <SENDER_MIDDLE_NAME>S.</SENDER_MIDDLE_NAME> <SENDER_LAST_NAME>Karanja</SENDER_LAST_NAME> </Record> ... </RecordSet> </Activities>

我应该通过将带有状态代码的 XML 发送回通知服务器来做出响应。我看过这个例子和这个

但两者似乎都不能满足我的需求。我将如何解决这个问题，尤其是如何为传入的通知保持开放的管道。是否有为此量身定制的红宝石？