问题标签 [activemq-cpp]

我已经并行启动了 5 个调用特定函数的线程。如果其中一个返回某个特定值，我需要终止所有其他剩余线程的进一步处理，并且在onMessage()方法中异步调用特定函数。

我对 activemq 相对较新，我尝试做的第一件事就是从服务器进程发布到 5,000 个主题（每个股票一个主题）。服务器和经纪人设法跟上。

然而，在消费者方面，它非常奇怪。如果我使用一个通配符使用者（“mytopic.>”）订阅所有 5k 个主题，一切都会保持正常。但是，如果我尝试为每个主题订阅一个消费者，性能就会下降并且无法跟上。

我尝试过使用预取限制和优化的确认模式，似乎没有任何帮助。
知道为什么单个通配符能够在 5k 个单独的主题中表现不佳吗？我也可以自己解复用消息，但希望 activemq 能够尽可能高效地为我做这件事。

编辑：有关此的更多信息和更新：

• 我在每秒发布一次的 ~ 6,000 个主题上对此进行测试
• 我正在使用 activemq-cpp c++​​ 库，并且正在为所有主题创建 1 个会话。事实证明，activemq 实现非常低效，它在向会话传递消息时对每条消息的所有主题进行线性扫描（实际上是两次）。
• 更糟糕的是，如果您为每个主题创建一个会话，它会尝试为每个会话创建一个线程，以便很快爆发。
• 可是等等！连接上有一个选项 setAlwaysSessionAsync，因此会话不会创建自己的线程，太棒了！
• 哦！不是那么快，会话仍然会在非异步模式下创建一些 RW 互斥体，并且它们使用一些本土 TLS 数据，这些数据的硬编码限制为每个线程约 300 个实例......呃
• 好的，所以我不得不将我可以创建的会话数限制为 ~ 150（我猜其他对象也在使用 TLS 数据），然后在这些主题上循环我的主题......
• 如果我能控制多少线程可以处理数据外线会很好，但可惜那也没有暴露出来……呃，在activemq-cpp代码中硬编码

TLDR；activemq 是一堆乱七八糟的便便

使用侦听器设置 CMS 消费者涉及两个单独的调用：首先，获取消费者：

然后，在消费者上设置一个监听器：

如果实现在侦听器被激活之前订阅目的地（并从代理/路由器接收消息），消息是否会丢失？或者这些消息是否在内部排队并在激活时传递给侦听器？

为什么没有 API 调用以侦听器作为构造参数来创建使用者？（是因为 JMS 规范没有吗？）

（附录：这可能是 API 本身的一个缺陷。更合乎逻辑的顺序是从会话中实例化消费者，并cms::Consumer::subscribe( cms::Destination*, cms::MessageListener* )在 API 中有一个方法。）

我有 2 个带有队列的 activeMQ 代理，我希望能够记录传入的 ActiveMQ 消息代理远程地址。

在我执行拉消费者接收后： pull_consumer->receive(1000);

消息到达后，我这样做： ActiveMQConnection *amqconn = dynamic_cast<ActiveMQConnection*>(connection);

然后尝试获取远程地址： amqconn->getTransport().getRemoteAddress().c_str()但它总是空的

我检查了getTransport().isClosed()并且getTransport().isConnected()它们没有关闭和连接

还是有另一种方法来检索此连接信息？

我正在使用 ActiveMQ 5.10.2，并且正在尝试使用 Apache ActiveMQ CMS (activemq-cpp-library-3.9.5) 使用 C++ 客户端测试故障转移方案。两个代理之一的配置是：

客户的网址是：failover:(tcp://broker-1:61636,tcp://broker-2:616137)?reconnectSupported=true&timeout=3000

在生产者开始向消费者发送消息后，我杀死了 broker-1，希望看到到 broker-2 的转换。而不是这个，我收到以下错误：

我的问题是，如果问题出在onException方法上，还是故障转移机制的配置有问题。

亲切的问候

我正在使用 INDIVIDUAL_ACKNOWLEDGE 模式接收消息并使用通过 Openwire 协议实现 JMS 标准的 C++ CMS 库手动确认从 AMQ 接收到的消息。我有两个消费者（消费者 1，消费者 2）从同一个队列接收。当 Consumer1 收到消息时，Consumer 确认消息。当我杀死消费者 1 时，消费者 2 会收到消息。我没想到 Consumer2 会收到它，因为它已经被确认了。看起来我对 acknowledge() 的调用被忽略了。

任何线索或指示都会有所帮助。尝试连接到 ActiveMQ 5.12.1 或 Red Hat AMQ 7.2.1，同样的行为表明问题不在于 Messaging Broker 而是我这边。我对 acknowledge() 的调用被忽略了。

很高兴提供更多信息。

这是消息接收器调用：

定义：

留言电话：

确认呼叫：

消费者 1 输出：

杀了它。

消费者2输出：

我在这里从 Apache 获取示例代码：https ://activemq.apache.org/components/cms/example

（特别是生产者部分）并尝试重写它，因此它不会创建任何用于生产的线程。相反，在我的程序的主线程中，创建一个生产者对象并设置连接、会话、目的地等。然后它使用消息生产者发送消息。这一切都是在一个单例中完成的，因此我的程序只有一个 Producer 对象，并且只要它需要将任何消息转储到我的队列之一就可以访问它。这个示例代码似乎为每个线程创建了一个生产者，每次都设置它，只是为了发送一条消息，然后删除所有内容。每次你想从你的程序中产生一些东西时，它都会这样做。

当我尝试使用任何给定消息对消息生产者调用 send 时，我正在崩溃。经过一番挖掘后，我发现在发送调用之后它会尝试锁定互斥锁并进入临界区。我想这是为了线程？我的代码中根本不使用线程，所以我猜它会因此而崩溃......有谁知道绕过这个的方法？我不想使用多个线程，我不需要担心两个线程试图同时调用 send 或者使用互斥锁试图解决的任何问题。

我正在寻找示例或资源来证明对 ActiveMQ 作为资源管理器之一的支持，而 Tuxedo 作为 XA 事务管理器。我正在构建一个 C++ 应用程序来做同样的事情。我无法在 Tuxedo 社区或 Google 上找到相同的任何文档。