当前位置: 首页 > 知识库问答 >
问题:

正在尝试测试队列持久性。在消费者上设置断点时无法发送消息

夹谷成仁
2023-03-14

我正在尝试测试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>

共有3个答案

顾宣
2023-03-14

activemq代理将等待几秒钟,以便客户端在强制断开连接之前发送wire格式。在连接URL上,尝试添加以下参数,以将该时间延长到允许您进行调试的时间:

tcp://localhost:61616?wireFormat.maxInactivityDurationInitalDelay=30000 

整数值是等待的毫秒数。

茅才
2023-03-14

"java.io.IOException: Wire格式协商超时:对等方没有发送他的线格式"看起来很清楚。您正在阻止作为网络连接另一端的客户端线程。服务器正在尝试与客户端交互时获得网络超时。网络连接是一种很难通过任意挂起线程来调试的情况。

东门俊智
2023-03-14

我发现了问题!

我在嵌入式ActiveMQ使用者上设置的远程断点是一个默认断点,suspend policty=all。

由于消费者和服务器在同一个JVM上运行,我也暂停了所有ActiveMQ服务器线程!

解决方案是使用断点挂起策略=线程,这样只有消费者线程被挂起,服务器线程才能继续运行。

 类似资料:
  • 如何修复ProbableAuthenticationError? 主机是Debian7和Python 2.7.3和pika 0.9.14,来宾是Ubuntu 15.04和rabbitmq-server 3.4.3-2

  • 我在ActiveMQ中使用异步消息使用者。我的制作人工作正常,向队列发送消息。现在,我的异步消息消费者正在等待调用onMessage(),但这从未发生过。因此,问题是: 异步使用者不会使用消息 ActiveMQ日志的快照还显示了许多刚刚堆积在挂起状态中的消息: 我想不出问题到底出在哪里。 计数: toPageIn 78 只是不断增加,信息仍然无法传递给消费者。 是服务器端问题还是客户端问题?

  • 我目前正在开发Kafka模块,我正在使用Kafka通信的抽象。我能够集成生产者 Spring Boot测试类 监听器类 我的问题是:在测试类中,我断言分区、有效负载等是从BlockingQueue轮询的,然而,我的问题是如何验证用KafkaListener注释的类中的业务逻辑是否得到正确执行,并根据错误处理和其他业务场景将消息路由到不同的主题。在一些示例中,我看到了CountDownLatch的断

  • 我正在使用Spring Kafka consumer,它从主题中获取消息并将其保存到数据库中。如果满足故障条件,例如db不可用,kafka消费者库是否提供重试机制?如果是,是否有方法设置不同的重试间隔,如第1次重试应在5分钟后进行,第2次重试应在30分钟后进行,第3次重试应在1小时后进行等。

  • 使用RabbitMQ,有没有一种方法可以将消息从队列“推送”给使用者,而不是让使用者从队列“轮询并拉出”消息? 这也是我目前正在进行的一个项目引起一些争论的原因。一个方面的论点是,让使用者(即windows服务)“轮询”队列直到新消息到达,与将消息从队列自动“推送”到订户/使用者的想法相比,这种想法有些低效,也不太理想。 我似乎只能找到支持消费者从队列中“轮询并拉出”消息的信息(例如,使用wind

  • 问题内容: 我试图将字符串消息发送到在weblogic服务器中创建的JMS队列中。我使用Eclipse IDE,当我运行Web应用程序时,出现以下错误,tomcat服务器关闭。错误是 请帮助我。谢谢和最诚挚的问候 问题答案: 基于对该问题的一些快速研究,它似乎是由于在应用服务器和客户端之间使用不同的JDK级别引起的。我看到的大多数示例都表明,在Java 5上运行Weblogic时在客户端上使用Ja