- Java 双重比较
- java - 比较器与 Apache BeanComparator
- Objective-C 完成 block 导致额外的方法调用?
- database - RESTful URI 是否应该公开数据库主键?
我正在尝试测试 ActiveMQ 的队列持久性。
我有一个带有唯一消费者的嵌入式 ActiveMQ 服务器。此嵌入式服务器从许多其他 JVM 应用程序接收 JMS 消息。
它工作正常,消费者应用程序收到通知。
所以我尝试测试消息的持久性。我在消费者的 MessageListener 上放置了一个(远程)断点,这样我就可以将许多消息排入队列并使 ActiveMQ 服务器崩溃。在服务器重新启动时,我希望所有排队的消息都能够被使用,而不是丢失。
然后我尝试了那个测试。我在第一条消息发送时进入了那个断点。但是对于我尝试发送的所有消息,我在生产者端得到以下堆栈跟踪:
Exception in thread "main" org.springframework.jms.UncategorizedJmsException: Uncategorized exception occured during JMS processing; nested exception is javax.jms.JMSException: Wire format negotiation timeout: peer did not send his wire format.
at org.springframework.jms.support.JmsUtils.convertJmsAccessException(JmsUtils.java:316)
at org.springframework.jms.support.JmsAccessor.convertJmsAccessException(JmsAccessor.java:168)
at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:469)
at org.springframework.jms.core.JmsTemplate.send(JmsTemplate.java:534)
at org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:612)
at org.springframework.jms.core.JmsTemplate.convertAndSend(JmsTemplate.java:604)
at com.xxxxxxxxxxx.mobilepush.client.RealClientTest.main(RealClientTest.java:29)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at com.intellij.rt.execution.application.AppMain.main(AppMain.java:120)
Caused by: javax.jms.JMSException: Wire format negotiation timeout: peer did not send his wire format.
at org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:62)
at org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1380)
at org.apache.activemq.ActiveMQConnection.ensureConnectionInfoSent(ActiveMQConnection.java:1466)
at org.apache.activemq.ActiveMQConnection.createSession(ActiveMQConnection.java:308)
at org.springframework.jms.support.JmsAccessor.createSession(JmsAccessor.java:196)
at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:457)
... 9 more
Caused by: java.io.IOException: Wire format negotiation timeout: peer did not send his wire format.
at org.apache.activemq.transport.WireFormatNegotiator.oneway(WireFormatNegotiator.java:98)
at org.apache.activemq.transport.MutexTransport.oneway(MutexTransport.java:68)
at org.apache.activemq.transport.ResponseCorrelator.asyncRequest(ResponseCorrelator.java:81)
at org.apache.activemq.transport.ResponseCorrelator.request(ResponseCorrelator.java:86)
at org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1351)
... 13 more
我不明白为什么当我的消费者处于我的断点时我的生产者会被阻止。
我的经纪人 uri 是:mobilepush.activemq.broker.transport.connector.uri=tcp://0.0.0.0:61616
生产者通过 tcp 连接到代理。与代理位于同一位置的消费者通过 vm://localhost
进行连接。
我的配置非常简单:
SERVER:
<!-- lets create an embedded ActiveMQ Broker -->
<amq:broker useJmx="false" persistent="true">
<amq:transportConnectors>
<amq:transportConnector uri="${mobilepush.activemq.broker.transport.connector.uri}" />
</amq:transportConnectors>
<amq:persistenceAdapter>
<amq:kahaPersistenceAdapter directory="${mobilepush.activemq.broker.queue.persistence.directory}" maxDataFileLength="100 Mb"/>
</amq:persistenceAdapter>
</amq:broker>
CONSUMER:
(management namespace and xebia class it only a JMX decorator)
<bean id="connectionFactory" class="fr.xebia.management.jms.SpringManagedConnectionFactory">
<property name="connectionFactory">
<bean class="org.apache.activemq.ActiveMQConnectionFactory" >
<property name="brokerURL" value="${mobilepush.activemq.broker.uri}"/>
</bean>
</property>
</bean>
<bean id="pushConsumer" class="com.xxxxxxxxxxxxxxx.mobilepush.messaging.jms.PushConsumer">
<property name="jmsPushMessageConverter" ref="jmsPushMessageConverter"/>
<property name="pushDelegate" ref="directPushDelegate"/>
</bean>
<management:executor-service id="pushConsumerExecutor"
pool-size="${mobilepush.consumer.thread.min}-${mobilepush.consumer.thread.max}" keep-alive="60" />
<jms:listener-container
task-executor="pushConsumerExecutor"
connection-factory="connectionFactory"
acknowledge="auto"
container-class="fr.xebia.springframework.jms.ManagedDefaultMessageListenerContainer">
<jms:listener destination="mobilepush.queue" ref="pushConsumer" method="onMessage" />
</jms:listener-container>
PRODUCER:
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" >
<property name="brokerURL" value="${mobilepush.activemq.broker.uri}"/>
</bean>
<bean id="mobilePushJmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="defaultDestination" ref="mobilePushQueue"/>
<property name="messageConverter" ref="jmsPushMessageConverter"/>
<property name="connectionFactory">
<!-- lets wrap in a pool to avoid creating a connection per send -->
<bean class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory">
<ref local="connectionFactory" />
</property>
</bean>
</property>
</bean>
最佳答案
我找到问题了!
我在嵌入式 ActiveMQ 消费者上设置的远程断点是一个带有 suspend-policty=all 的默认断点。
由于消费者和服务器在同一个 JVM 上运行,我还暂停了所有 ActiveMQ 服务器线程!
解决方案是使用断点suspend-policy=thread,这样只有消费者线程被挂起,服务器线程可以继续运行。
关于java - 尝试测试队列持久性。消费者断点时无法发送消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/12267168/
kafka的Java客户端-消费者 一、kafka消费方式 pull(拉)模式:consumer采用从broker中主动拉取数据。Kafka 采用这种方式 push(推)模式:Kafka没有采用这种方
我编写这个小应用程序是为了解决 Python 中的经典生产者/消费者问题。我知道我可以使用线程安全的队列机制来解决这个问题,但我有兴趣自己解决这个问题来学习。 from threading impor
下面是一个示例消费者/生产者模型的代码: int buffer[MAX]; int fill_ptr = 0; int use_ptr = 0; int count = 3; void put(int
我的消费者、生产者程序有问题,它似乎可以加载,但返回段错误。我已经尝试了一切来修复它,但仍然失败!将不胜感激任何帮助。笔记;代码真的很多,semaphore.h的代码都在里面,有谁想测试一下。其余代码
我正在阅读著名的操作系统概念书(Avi Silberschatz、Peter Baer Galvin、Greg Gagne)第 9 版:http://codex.cs.yale.edu/avi/os-
我正在尝试构建一个服务,为许多异步客户端提供队列以发出请求并等待响应。我需要能够通过每 Y 个持续时间的 X 个请求来限制队列处理。例如:每秒 50 个 Web 请求。它用于第 3 方 REST 服务
我正在尝试使用一组资源来实现生产者/消费者模式,因此每个线程都有一个与之关联的资源。例如,我可能有一个任务队列,其中每个任务都需要一个 StreamWriter写出它的结果。每个任务还必须有参数传
为什么我们需要 Azure 存储帐户上的 blob 容器用于 Eventhub 消费者客户端(我使用的是 python)。为什么我们不能像在 Kafka 中那样直接使用来自 Eventhub(Kafk
我有一个有趣的生产者-消费者衍生产品需要实现,但我无法理解它的算法。因此,每个生产者都会“产生”给定范围(最小值,最大值)之间的数字,这对除以给定“商”给出了相同的提醒。对于消费者来说也是如此。 额外
我需要实现一种生产者/消费者方案,出于性能原因,消费者尝试在一批中处理许多工作项(每个工作项都会耗尽工作队列)。 目前,我只是创建固定数量的相同工作人员,它们在循环中的同一队列上工作。由于其中一些可能
为什么我们需要 Azure 存储帐户上的 blob 容器用于 Eventhub 消费者客户端(我使用的是 python)。为什么我们不能像在 Kafka 中那样直接使用来自 Eventhub(Kafk
我的关系必须按如下方式运作;线程 A 向线程 B 发布一些更改,线程 B 接受该更改并将其发布到线程 C。 问题是生产者-消费者,我使用 BlockingQueue 仅用两个实体来实现它没有问题。我怎
我一直在研究 PC 问题,以了解 Java 同步和线程间通信。使用底部的代码,输出为 Producer produced-0 Producer produced-1 Producer produced
我编写了代码来实现生产者-消费者问题,它似乎工作正常,不需要同步。这可能吗? 如何测试代码并检查它是否确实正常工作?我如何知道是否会发生死锁?现在,我没有跳出循环(即生产者不断插入,消费者不断在无限循
我必须完成一项练习,我必须使用至少一个生产者线程和 x 个消费者线程的生产者/消费者模式在我的文件夹路径中查找“.java”文件。 生产者消费者级:首先,当生产者完成查找文件时,我尝试通过设置从 tr
我被分配了一项类(class)作业来实现消费者/生产者问题的解决方案,该解决方案使用单个生产者、单个消费者和循环缓冲区。这应该用 C 语言编写。 不幸的是,我们没有获得任何学习 Material ,并
有人可以检查我的代码并告诉我是否走在正确的轨道上。我似乎有点迷失了。如果您看到我的错误,请告诉我它们。 我想做的是使用我自己的信号量以及 GCD 来解决有界缓冲区问题。 提前致谢.. sema.c v
我要处理有界缓冲区、生产者消费者问题,只能修改 prod 和 cons 函数。此代码仅在一个消费者和生产者线程上运行,不会出现任何问题。但对于每个都有多个,迟早总会给我带来同样的问题: p5p1:
我有一个从多个线程访问的类的实例。此类接受此调用并将元组添加到数据库中。我需要以串行方式完成此操作,因为由于某些数据库约束,并行线程可能会导致数据库不一致。 由于我不熟悉 C# 中的并行性和并发性,所
我正在尝试编写一个批量邮件服务,它有两种方法: add(Mail mail):可以发送邮件,由Producers调用 flushMailService():刷新服务。消费者应该获取一个列表,并调用另一
我是一名优秀的程序员,十分优秀!