问题标签 [consumer]

我一直在尝试使用渠道来构建某种生产者/消费者。我有一个requests渠道，许多生产者推送请求，然后我有processRequests处理请求的渠道。

我想知道的是，一旦请求得到满足，将响应发送回生产者（以及正确的，因为不止一个）的最佳方式是什么？基本上如何使它成为一个双向通道？

知道怎么做吗？

我正在使用来自 Kafka 站点的 ConsumerGroupExample 代码测试 Kafka 高级消费者。我想检索 Kafka 服务器配置中名为“test”的主题的所有现有消息。查看其他博客，auto.offset.reset 应该设置为“最小”才能获取所有消息：

我真正遇到的问题是：高级消费者的等效 Java api 调用是什么，相当于：

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

我有多个生产者，每个生产者都将数据推送到自己的阻塞队列中。每个生产者的数据都是独立处理的（因此有单独的队列）。

我的单个消费者目前轮询每个队列，每个队列都有超时。

程序运行良好，但我正在尝试优化，我不喜欢“轮询”。这是解决此问题的最有效方法吗？

我使用每个生产者的消费者编写了相同的应用程序，但情况稍差。

谢谢 。

谁能告诉我是否有一个属性允许配置 JMS 消费者代理轮询队列/主题的频率？

我似乎无法通过 RabbitMQ 让 Castle Windsor Integration 为 Mass Transit 工作。在我将温莎引入图片之前，一切都很好。我引用了 Castle.Windsor 3.2 和 MassTransit.WindsorIntegration 2.9 并配置了容器以在我的应用程序中使用。我正在通过以下方式注册 MassTransit 消费者：

当我在运行此行后调试和检查容器时，我可以看到它成功注册了所有消费者以及我的所有其他组件。然后我有以下代码来初始化和注册服务总线：

我正在使用LoadFrom(IWindsorContainer container)MassTransit.WindsorIntegration 提供的扩展方法。

到目前为止，我发现的所有示例都停在这里，并表明这是您应该做的所有事情。对我来说不幸的是，我的消费者从未被调用，消息只是堆积在队列中（最终超时并进入错误队列）。消息完全显示在消费者队列中的事实（+ 我看到一个消费者通过 RabbitMQ 管理工具绑定到队列）向我表明消费者可能被正确订阅 - 所以我不确定在哪里问题出在。

我为 Windsor 和 MassTransit 添加了 NLog 日志记录，但日志中没有显示任何错误。我不确定此时我应该如何进行故障排除。有任何想法吗？

此外，这个应用程序目前只是一个使用 Topshelf 进行开发的控制台应用程序。最终它将作为 Windows 服务安装。不确定这是否相关，但我想我会提到它以防万一。

更新

作为一个测试，我创建了一个非常简单的消费者，它带有一个无参数的构造函数来处理一个非常简单的测试消息。这个消费者被成功调用！然而，“真正的”消费者具有需要通过构造函数注入它们的依赖项。我希望 Container 能解决这些问题，但显然它遇到了一些麻烦。奇怪的是，日志中没有显示任何关于它的内容。敬请关注...

我一直在尝试使用 ruby​​ 消费者非常简单地扇出这些队列，它只是订阅该交换/队列并接收消息。现在每当发布新消息时都会出现问题。他们没有收到消息，这意味着他们没有消费，也没有列出任何消费者。一旦您再次将队列与交换重新绑定并重新启动 ruby​​ 应用程序，它就会再次开始使用。然后它又回到了lembo！有时，当您重新启动 ruby​​ 应用程序几次时，它会起作用。任何想法？

用于以下消费者的代码：

我正在通过不同的方法解决生产者/消费者问题。我目前正在研究的一个涉及我自己的信号量实现，与 Peterson 的解决方案配对以保护信号量的 down() 和 up() 函数调用。

问题是，我正在使用 C 语言工作，我能想到的创建信号量的唯一方法是使用带有函数指针的结构，用于 down() 和 up()。这是正确的想法吗？

我正在实现我自己的自定义组件，我发现我需要为消费者提供两个用例：

• 第一个将尝试每隔一段时间获取 N 条可用消息（轮询消费者）
• 第二个是订阅者消费者，当消息可用时获取消息。

我的主要问题是是否可以实现这两种类型。我一直在尝试编写一些代码，但似乎如果你正在开发一个 PollingConsumer你不能实现另一种类型。另外，如果可能的话，是否有任何关于如何执行此操作的示例、文章或指南？我一直在寻找它没有出现。

谢谢！

我编写了一个模拟生产者消费者问题的程序，我遇到了几个问题。这是使用 Win32 API 编写的。

我正在使用两个满和空的信号量来对存储项目的缓冲区执行计数。还有一个互斥锁来控制对临界区的访问。我有两个功能：一个基于简单的计算创建一个项目：线程 * 1000000 + 计数，而另一个消耗它。

缓冲区中有 10 个空格，但是程序应该希望能够为不同数量的空格和线程工作。生产者线程通过缓冲区然后重新开始，直到信号量达到 10。我遇到了两个问题，我似乎也找不到解决方案，也没有在调试器中获得很多细节。

1）具有 printf() 函数的注释部分每次都会使线程崩溃。没有任何东西被打印到控制台。我也尝试过只使用一个字符串，而没有输出其他变量。没有成功。我发现 printf() 在多线程环境中使用时可能会遇到问题的文档，但是在这种情况下，它位于关键部分内，并且在第一次尝试时会崩溃。如何将该信息打印到控制台？

2）当我删除打印语句并运行代码时，线程每次仍然在不同的点崩溃。我不知道为什么。生产者线程应该等待空信号量和互斥体，将项目放入，然后增加完整信号量的计数（表示已添加项目）。当它达到 10 时，它应该停止。消费者线程等待完整的信号量和互斥量，删除一个项目，然后增加空信号量的计数。我写的方式有什么导致这些线程退出的吗？

这就是我得到的：

程序“[5348] OperatingSystem.exe”已退出，代码为 0 (0x0)。

我正在使用 RabbitMQ，并且我有一个包含电子邮件消息的队列。我的消费者服务使消息出队并尝试发送它们。如果出于任何原因，我的消费者无法发送消息，我想将消息重新排队以再次发送。我意识到我可以做一个 basicNack 并将 requeue 标志设置为 true，但是，我不想无限期地重新排队消息（例如，如果我们的电子邮件系统出现故障，我不想不断地重新排队未发送的消息）。我想定义一个有限的次数，我可以将消息重新排队以再次发送。但是，当我将其出列并发送 nack 时，我无法在电子邮件对象上设置字段。队列中的消息上不存在更新的字段。还有其他方法可以解决这个问题吗？提前致谢。