首页 文章

试图测试队列持久性 . 消费者断点时无法发送消息

提问于
浏览
4

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

4 回答

  • 0

    我发现了问题!

    我在嵌入式ActiveMQ使用者身上放置的远程断点是一个默认断点,其中suspend-policty = all .

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

    解决方案是使用断点suspend-policy = thread,以便只挂起使用者线程,并且服务器线程可以继续运行 .

  • 0

    "java.io.IOException: Wire format negotiation timeout: peer did not send his wire format" 似乎很清楚 . 您正在阻止作为网络连接另一端的客户端线程 . 服务器正在尝试与客户端进行交互的网络超时 . 网络连接是一种通过任意挂起线程很难调试的情况 .

  • 4

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

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

    整数值是要等待的毫秒数 .

  • 0

    我通过使用最新的logback-core和logback-classic jar文件修复了这个问题,该文件是1.1.2

相关问题