我的代码中有一个Spring JmsListener。它接收和消费消息2天,但突然在这2天后,它没有收到来自外部Activemq的消息。然而,它的队列中有一些挂起的消息。当我重置Activemq和消费者时,消费者会收到大量消息。当消息挂起时,连接到Activemq的消费者(代表Spring致动器日志)。日志和配置显示Activemq没有将消息推送给消费者。我有另一个像这个消费者一样的服务从其他队列接收消息。第二个消费者也有同样的问题。我想知道为什么会发生这个问题,我如何解决这个问题。这是我的配置和日志:
消费者:
应用xml:
spring.jms.pub-sub-domain=false
spring.jms.template.delivery-mode=persistent
spring.activemq.broker-url=${BROKER_URL:failover:(tcp://activemq1:61616,tcp://activemq2:61616)?maxReconnectDelay=2500}
spring.activemq.user=${BROKER_USER:admin}
spring.activemq.password=${BROKER_PASSWORD:admin}
JmsConfiguration:
@EnableJms
@Configuration
public class JmsConfiguration {
@Bean
public JmsListenerContainerFactory<?> myFactory(ActiveMQConnectionFactory connectionFactory,
DefaultJmsListenerContainerFactoryConfigurer configurer) {
DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
redeliveryPolicy.setMaximumRedeliveries(0);
connectionFactory.setRedeliveryPolicy(redeliveryPolicy);
factory.setMessageConverter(messageConverter());
configurer.configure(factory, connectionFactory);
return factory;
}
private MessageConverter messageConverter() {
MappingJackson2MessageConverter messageConverter = new MappingJackson2MessageConverter();
messageConverter.setTargetType(MessageType.TEXT);
messageConverter.setTypeIdPropertyName("_type");
return messageConverter;
}
}
听众:
@Component
public class MessageReceiver {
@JmsListener(destination = Constant.OFFICE_REQUEST_QUEUE, containerFactory = "myFactory")
public void receive(RequestMessage requestMessage, @Headers Map<String,Object> headers) throws NonPersistenceServiceException {
// do someting on received message
}
}
Activemq配置:
activemq.xml:
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">
<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
<property name="locations">
<value>file:${activemq.conf}/credentials.properties</value>
</property>
</bean>
<bean id="logQuery" class="io.fabric8.insight.log.log4j.Log4jLogQuery"
lazy-init="false" scope="singleton"
init-method="start" destroy-method="stop">
</bean>
<bean id="oracleDS" class="oracle.jdbc.pool.OracleDataSource" destroy-method="close">
<property name="URL" value="jdbc:oracle:thin:@(DESCRIPTION= (SDU=32768)(ADDRESS=(PROTOCOL=TCP)(HOST= dms-db1-vip)(PORT=1521)) (ADDRESS=(PROTOCOL=TCP)(HOST= dms-db2-vip)(PORT=1521)) (LOAD_BALANCE=yes)(FAILOVER=ON)(CONNECT_DATA=(SERVER=DEDICATED)(SERVICE_NAME=orcl)(failover_mode=(type=select)(method=basic)(retries=5)(delay=1))))"/>
<property name="user" value="ora_user"/>
<property name="password" value="ora_pass"/>
</bean>
<broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}" >
<plugins>
<redeliveryPlugin fallbackToDeadLetter="true"
sendToDlqIfMaxRetriesExceeded="true">
<redeliveryPolicyMap>
<redeliveryPolicyMap>
<defaultEntry>
<redeliveryPolicy
useExponentialBackOff="true"
backOffMultiplier="2"
maximumRedeliveryDelay="1200000"
maximumRedeliveries="82"/>
</defaultEntry>
</redeliveryPolicyMap>
</redeliveryPolicyMap>
</redeliveryPlugin>
</plugins>
<destinationPolicy>
<policyMap>
<policyEntries>
<policyEntry queue=">">
<deadLetterStrategy>
<individualDeadLetterStrategy queuePrefix="DLQ." useQueueForQueueMessages="true"/>
</deadLetterStrategy>
</policyEntry>
</policyEntries>
</policyMap>
</destinationPolicy>
<managementContext>
<managementContext createConnector="false"/>
</managementContext>
<persistenceAdapter>
<jdbcPersistenceAdapter dataSource="#oracleDS" />
</persistenceAdapter>
<systemUsage>
<systemUsage>
<memoryUsage>
<memoryUsage percentOfJvmHeap="70" />
</memoryUsage>
<storeUsage>
<storeUsage limit="100 gb"/>
</storeUsage>
<tempUsage>
<tempUsage limit="50 gb"/>
</tempUsage>
</systemUsage>
</systemUsage>
<transportConnectors>
<transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="amqp" uri="amqp://0.0.0.0:5672?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="stomp" uri="stomp://0.0.0.0:61613?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="mqtt" uri="mqtt://0.0.0.0:1883?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
<transportConnector name="ws" uri="ws://0.0.0.0:61614?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
</transportConnectors>
<shutdownHooks>
<bean xmlns="http://www.springframework.org/schema/beans" class="org.apache.activemq.hooks.SpringContextHook" />
</shutdownHooks>
</broker>
<import resource="jetty.xml"/>
</beans>
以下是耗电元件中执行器的日志:
{"log":"20:20:26.775 @@@ [ActiveMQ Task-1] INFO o.a.activemq.transport.failover.FailoverTransport @@@ @@@ @@@\r\n","stream":"stdout"}
{"log":" Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:36.638 @@@ [ActiveMQ Task-1] INFO o.a.activemq.transport.failover.FailoverTransport @@@ @@@ @@@\r\n","stream":"stdout"}
{"log":" Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:36.779 @@@ [ActiveMQ Task-1] INFO o.a.activemq.transport.failover.FailoverTransport @@@ @@@ @@@\r\n","stream":"stdout"}
{"log":" Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:46.464 @@@ [ActiveMQ Task-1] INFO o.a.activemq.transport.failover.FailoverTransport @@@ @@@ @@@\r\n","stream":"stdout"}
{"log":" Successfully connected to tcp://activemq1:61616\r\n","stream":"stdout"}
{"log":"20:20:46.774 @@@ [ActiveMQ Task-1] INFO o.a.activemq.transport.failover.FailoverTransport @@@ @@@ @@@\r\n","stream":"stdout"}
Activemq错误日志:
2019-03-05 20:19:55,448 | WARN | Transport Connection to: tcp://10.42.1.0:63493 failed: java.io.IOException: Frame size of 1 GB larger than max allowed 100 MB | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///10.42.1.0:63493@61616
2019-03-05 20:20:05,776 | WARN | Transport Connection to: tcp://10.42.1.0:63498 failed: java.io.EOFException | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///10.42.1.0:63498@61616
2019-03-05 20:20:05,776 | WARN | Transport Connection to: tcp://10.42.1.0:63497 failed: java.io.EOFException | org.apache.activemq.broker.TransportConnection.Transport | ActiveMQ Transport: tcp:///10.42.1.0:63497@61616
这种情况通常是两个问题之一的结果:
>
某些网络组件(路由器、防火墙)由于不活动而自动关闭套接字,而不通知客户端或服务器,因此,除非启用心跳,否则连接将失效。
消费者线程不知何故“卡在”用户代码中,因此它停止接收新消息;获取堆栈转储以查看容器线程正在做什么。
我们面临着ActiveMQ及其消费者的随机问题。我们观察到,很少有消费者不接收消息,即使他们连接到ActiveMQ队列。但在消费者重启后,它工作正常。 我们在ActiveMQ端有一个名为testQueue的队列。消费者正试图将消息从该队列中解列。为此,我们正在使用Spring的DefaultMessageListenerContainer。消息正在从ActiveMQ代理传递到使用者节点。从tcpd
我有一个JMS生产者和一个消费者,代理是ActiveMQ,参考下面的代码: 寄件人代码 接收码 问题是 ActiveMQ 队列无法接收来自发送方的消息(请参阅屏幕截图): 当我从 Web 控制台发送消息时,该消息在队列中收到,但来自创建者的消息不会进入队列。 另一个有趣的行为是(如队列接收器代码中所示,接收器在收到第一条消息后退出),同样,当我启动接收器时,它会收到相同的消息,并继续执行,直到我关
如果我创建上面的类并尝试在tomcat7上部署war,我会看到以下错误。
因此,我使用Spring integration链接JMS和ActiveMQ,如下所示:- 如何使其工作,以便发送到此队列并从中接收消息?请帮忙。
我很感激你在这方面的帮助。 我正在构建一个ApacheKafka消费者,以订阅另一个已经运行的Kafka。现在,我的问题是,当我的制作人将消息推送到服务器时。。。我的消费者没有收到。。我在打印的日志中得到以下信息: 我不确定我是否遗漏了任何重要的配置。。。但是,我可以使用WireShark看到一些来自我的服务器的消息,但是我的消费者没有消费这些消息。。。。 我的代码是示例消费者示例的精确副本:ht
本文向大家介绍python 发送和接收ActiveMQ消息的实例,包括了python 发送和接收ActiveMQ消息的实例的使用技巧和注意事项,需要的朋友参考一下 ActiveMQ是java开发的消息中间件服务。可以支持多种协议(AMQP,MQTT,OpenWire,Stomp),默认的是OpenWire。而python与ActiveMQ的通信使用的是Stomp协议。而如果你的服务没有开启则需要配