ActiveMQ消息去排队但未被消耗

Question

我有一个JBoss web应用程序，该应用程序目前正在使用嵌入式HornetQ进行JMS。我们想切换到一个ActiveMQ HA集群，但是我遇到了一些奇怪的问题。我的一个队列(periodicDerivationQueue)没有像对HornetQ那样运行。AMQ控制台显示消息是排队的和退出队列的，但是它们并没有对我的使用者产生影响。起初，我认为消息是出于某种原因而排到DLQ中的，但情况似乎并非如此。据我所知，AMQ不会创建DLQ，除非它是必要的。当我看经纪人的时候，没有DLQ。我怎样才能知道我的信息要去哪里？

由于反射，我在堆栈的应用程序端调试也有困难。我想在AMQ方面设置一个断点，看看我的消息发生了什么，但我不知道该放在哪里。这里有什么想法吗？

这可能是序列化问题吗？我听说有时JMS代理之间序列化的差异会导致奇怪的行为。

我真的被困在这里了，任何帮助都将不胜感激。请参阅下面的配置信息。

野蝇8.2

AMQ 5.13

消费者(消息不在此制作)

public class PeriodicDerivationExecutionHandlerImpl implements PeriodicDerivationExecutionHandler {

protected DerivationService derivationService;
protected DerivationModelService derivationModelService;

protected Logger logger = LoggerFactory.getLogger(this.getClass());

@Override
public void executeDerivation(PeriodicDerivation params) throws Exception{

    JbpmHibernateUtil.openSession();
    Derivation derivation = null;
    try{            

        if (params.isGroup()){
            derivation = new GroupDerivation();

            GroupQueryParameters qp = new GroupQueryParameters();
            qp.setGroupName(params.getItemName());
            derivation.setDerivedItem(derivationModelService.getGroup(qp));

        }else{
            derivation = new DeterminantDerivation();

            DeterminantQueryParameters qp = new DeterminantQueryParameters();
            qp.setDeterminantName(params.getItemName());            
            derivation.setDerivedItem(derivationModelService.getDeterminant(qp));

        }

        logger.info("Executing periodic derivation [" + derivation + "]");

        derivation.setModelEffectiveDate(new DateTime());
        derivation.setPeriod(params.getPeriod());
        derivation.getProcessParameters().add(new DerivationProcessParameter(PeriodicDerivation.PERIODIC_PROCESS_VAR, true));
        derivation.setExecutionMode(DerivationExecutionMode.SYNCHRONOUS_LOCAL);
        derivationService.executeDerivation(derivation);

        JbpmHibernateUtil.closeSession(true);
    }catch(Exception e){
        logger.error("Periodic derivation execution failed for [" + derivation + "]",e);
        JbpmHibernateUtil.closeSession(false);
        throw new Exception("Periodic derivation execution failed for [" + derivation + "]",e);
    }
}

public DerivationService getDerivationService() {
    return derivationService;
}

public void setDerivationService(DerivationService derivationService) {
    this.derivationService = derivationService;
}

public DerivationModelService getDerivationModelService() {
    return derivationModelService;
}

public void setDerivationModelService(DerivationModelService derivationModelService) {
    this.derivationModelService = derivationModelService;
}

}

消费者XML配置

<int:gateway id="periodicDerivationExecutionGateway"
        service-interface="com.etse.jbpm.scheduler.PeriodicDerivationExecutionHandler">
        <int:method name="executeDerivation" request-channel="periodicDerivationChannel" />
    </int:gateway>

    <bean id="periodicDerivationExecutor"
        class="com.etse.jbpm.scheduler.PeriodicDerivationExecutionHandlerImpl">
        <property name="derivationService" ref="derivationService" />
        <property name="derivationModelService" ref="derivationModelService" />
    </bean>

    <int:service-activator input-channel="periodicDerivationChannel"
        ref="periodicDerivationExecutor" method="executeDerivation" />

    <int-jms:channel id="periodicDerivationChannel"
        queue-name="${jms.destination.name.periodicderivation}" concurrency="${integration.listener.threads.maximum}"
        task-executor="periodicDerivationTaskExecutor" />

ActiveMQ Standalone.xml (Jboss)

<subsystem xmlns="urn:jboss:domain:resource-adapters:2.0">
        <resource-adapters>
            <resource-adapter id="activemq-rar.rar">
                <archive>
                    activemq-rar.rar
                </archive>
                <transaction-support>XATransaction</transaction-support>
                <config-property name="ServerUrl">
                    tcp://127.0.0.1:61616?jms.rmIdFromConnectionId=true
                </config-property>
                <config-property name="UserName">
                    admin
                </config-property>
                <config-property name="Password">
                    admin
                </config-property>
                <connection-definitions>

                    <connection-definition 
                       class-name="org.apache.activemq.ra.ActiveMQManagedConnectionFactory" 
                       jndi-name="java:/ConnectionFactory"  
           enabled="true" 
           pool-name="ConnectionFactory"> 
                        <xa-pool>
                            <min-pool-size>1</min-pool-size>
                            <max-pool-size>20</max-pool-size>
                            <prefill>false</prefill>
                            <is-same-rm-override>false</is-same-rm-override>
                        </xa-pool>
                        <recovery>
                            <recover-credential>
                                <user-name>admin</user-name>
                                <password>admin</password>
                            </recover-credential>
                            <recover-plugin class-name="org.jboss.jca.core.recovery.ConfigurableRecoveryPlugin">
                                <config-property name="EnableIsValid">
                                    false
                                </config-property>
                                <config-property name="IsValidOverride">
                                    true
                                </config-property>
                                <config-property name="EnableClose">
                                    true
                                </config-property>
                            </recover-plugin>
                        </recovery>
                    </connection-definition>

                </connection-definitions>

Queues/Topics

<admin-objects>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" 
                                jndi-name="java:jboss/exported/jms/queue/bpm/deferredBpmCommandQueue" 
                                use-java-context="true" 
                                pool-name="deferredBpmCommandQueue">
                        <config-property name="PhysicalName">
                            deferredBpmCommandQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" 
                                jndi-name="java:jboss/exported/jms/queue/bpm/asyncActionRequestQueue" 
                                use-java-context="true" 
                                pool-name="ActiveMQQueue.asyncActionRequestQueue">
                        <config-property name="PhysicalName">
                            asyncActionRequestQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" 
                                jndi-name="java:jboss/exported/jms/queue/bpm/DLQ" 
                                use-java-context="true" 
                                pool-name="DLQ">
                        <config-property name="PhysicalName">
                            DLQ
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:jboss/exported/jms/queue/bpm/cacheUpdateReplicationQueue" use-java-context="true" pool-name="ActiveMQQueue.cacheUpdateReplicationQueue">
                        <config-property name="PhysicalName">
                            cacheUpdateReplicationQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:jboss/exported/jms/queue/bpm/periodicDerivationQueue" use-java-context="true" pool-name="ActiveMQQueue.periodicDerivationQueue">
                        <config-property name="PhysicalName">
                            periodicDerivationQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:jboss/exported/jms/queue/bpm/asyncServiceSignalQueue" use-java-context="true" pool-name="ActiveMQQueue.asyncServiceSignalQueue">
                        <config-property name="PhysicalName">
                            asyncServiceSignalQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQTopic" jndi-name="java:jboss/exported/jms/queue/bpm/processEventTopic" use-java-context="true" pool-name="ActiveMQTopic.processEventTopic">
                        <config-property name="PhysicalName">
                            processEventTopic
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:jboss/exported/jms/queue/bpm/asyncActionReplyQueue" use-java-context="true" pool-name="ActiveMQQueue.asyncActionReplyQueue">
                        <config-property name="PhysicalName">
                            asyncActionReplyQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:jboss/exported/jms/queue/bpm/ExpiryQueue" use-java-context="true" pool-name="ActiveMQQueue.ExpiryQueue">
                        <config-property name="PhysicalName">
                            ExpiryQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQTopic" jndi-name="java:jboss/exported/jms/queue/bpm/asyncActionServiceStatusRequestTopic" use-java-context="true" pool-name="ActiveMQTopic.asyncActionServiceStatusRequestTopic">
                        <config-property name="PhysicalName">
                            asyncActionServiceStatusRequestTopic
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQTopic" jndi-name="java:jboss/exported/jms/queue/bpm/asyncActionAffinityRequestTopic" use-java-context="true" pool-name="ActiveMQTopic.asyncActionAffinityRequestTopic">
                        <config-property name="PhysicalName">
                            asyncActionAffinityRequestTopic
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:jboss/exported/jms/queue/bpm/jbpmJobQueue" use-java-context="true" pool-name="ActiveMQQueue.jbpmJobQueue">
                        <config-property name="PhysicalName">
                            jbpmJobQueue
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQTopic" jndi-name="java:jboss/exported/jms/queue/bpm/asyncActionAffinityReplyTopic" use-java-context="true" pool-name="ActiveMQTopic.asyncActionAffinityReplyTopic">
                        <config-property name="PhysicalName">
                            asyncActionAffinityReplyTopic
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQTopic" jndi-name="java:jboss/exported/jms/queue/bpm/cacheUpdateReplicationEventTopic" use-java-context="true" pool-name="ActiveMQTopic.cacheUpdateReplicationEventTopic">
                        <config-property name="PhysicalName">
                            cacheUpdateReplicationEventTopic
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQTopic" jndi-name="java:jboss/exported/jms/queue/bpm/asyncActionServiceStatusTopic" use-java-context="true" pool-name="ActiveMQTopic.asyncActionServiceStatusTopic">
                        <config-property name="PhysicalName">
                            asyncActionServiceStatusTopic
                        </config-property>
                    </admin-object>
                    <admin-object class-name="org.apache.activemq.command.ActiveMQQueue" jndi-name="java:jboss/exported/jms/queue/bpm/asyncActionServiceLogRecordQueue" use-java-context="true" pool-name="ActiveMQQueue.asyncActionServiceLogRecordQueue">
                        <config-property name="PhysicalName">
                            asyncActionServiceLogRecordQueue
                        </config-property>
                    </admin-object>
                </admin-objects>

Broker配置

<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>

<broker xmlns="http://activemq.apache.org/schema/core" brokerName="broker1" dataDirectory="${activemq.data}" persistent="true"> 

    <destinationPolicy>
        <policyMap>
          <policyEntries>
            <policyEntry topic=">" >

              <pendingMessageLimitStrategy>
                <constantPendingMessageLimitStrategy limit="1000"/>
              </pendingMessageLimitStrategy>
            </policyEntry>
          </policyEntries>
        </policyMap>
    </destinationPolicy>


    <managementContext>
        <managementContext createConnector="false"/>
    </managementContext>


    <persistenceAdapter>
      <kahaDB directory="${activemq.data}/kahadb"/> 
    </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&amp;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"/>

​

HornetQ Standalone.xml (Jboss)

<subsystem xmlns="urn:jboss:domain:messaging:2.0">
        <hornetq-server>
            <persistence-enabled>false</persistence-enabled>
            <jmx-management-enabled>true</jmx-management-enabled>
            <shared-store>true</shared-store>
            <journal-type>ASYNCIO</journal-type>
            <journal-file-size>102400</journal-file-size>
            <journal-min-files>2</journal-min-files>

            <connectors>
                <netty-connector name="netty" socket-binding="messaging"/>
                <netty-connector name="netty-throughput" socket-binding="messaging-throughput">
                    <param key="batch-delay" value="50"/>
                </netty-connector>
                <in-vm-connector name="in-vm" server-id="0"/>
            </connectors>

            <acceptors>
                <netty-acceptor name="netty" socket-binding="messaging"/>
                <netty-acceptor name="netty-throughput" socket-binding="messaging-throughput">
                    <param key="batch-delay" value="50"/>
                    <param key="direct-deliver" value="false"/>
                </netty-acceptor>
                <in-vm-acceptor name="in-vm" server-id="0"/>
            </acceptors>

            <security-settings>
                <security-setting match="#">
                    <permission type="send" roles="guest"/>
                    <permission type="consume" roles="guest"/>
                    <permission type="createNonDurableQueue" roles="guest"/>
                    <permission type="deleteNonDurableQueue" roles="guest"/>
                </security-setting>
            </security-settings>

            <address-settings>
                <address-setting match="#">
                    <dead-letter-address>jms.queue.DLQ</dead-letter-address>
                    <expiry-address>jms.queue.ExpiryQueue</expiry-address>
                    <redelivery-delay>0</redelivery-delay>
                    <max-size-bytes>104857600</max-size-bytes>
                    <page-size-bytes>10485760</page-size-bytes>
                    <page-max-cache-size>10</page-max-cache-size>
                    <address-full-policy>PAGE</address-full-policy>
                    <message-counter-history-day-limit>10</message-counter-history-day-limit>
                </address-setting>
            </address-settings>

            <jms-connection-factories>
                <connection-factory name="InVmConnectionFactory">
                    <connectors>
                        <connector-ref connector-name="in-vm"/>
                    </connectors>
                    <entries>
                        <entry name="java:/ConnectionFactory"/>
                    </entries>
                </connection-factory>
                <connection-factory name="RemoteConnectionFactory">
                    <connectors>
                        <connector-ref connector-name="netty"/>
                    </connectors>
                    <entries>
                        <entry name="java:jboss/exported/jms/RemoteConnectionFactory"/>
                    </entries>
                    <client-failure-check-period>30000</client-failure-check-period>
                    <connection-ttl>300000</connection-ttl>
                    <retry-interval>2000</retry-interval>
                    <retry-interval-multiplier>1</retry-interval-multiplier>
                    <max-retry-interval>2000</max-retry-interval>
                    <reconnect-attempts>100</reconnect-attempts>
                </connection-factory>
            </jms-connection-factories>

Queues/Topics

<jms-destinations>
                <jms-queue name="asyncActionRequestQueue">
                    <entry name="queue/bpm/asyncActionRequestQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/asyncActionRequestQueue"/>
                </jms-queue>
                <jms-queue name="asyncActionReplyQueue">
                    <entry name="queue/bpm/asyncActionReplyQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/asyncActionReplyQueue"/>
                </jms-queue>
                <jms-queue name="asyncServiceSignalQueue">
                    <entry name="queue/bpm/asyncServiceSignalQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/asyncServiceSignalQueue"/>
                </jms-queue>
                <jms-queue name="asyncActionServiceLogRecordQueue">
                    <entry name="queue/bpm/asyncActionServiceLogRecordQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/asyncActionServiceLogRecordQueue"/>
                </jms-queue>
                <jms-queue name="deferredBpmCommandQueue">
                    <entry name="queue/bpm/deferredBpmCommandQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/deferredBpmCommandQueue"/>
                </jms-queue>
                <jms-queue name="jbpmJobQueue">
                    <entry name="queue/bpm/jbpmJobQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/jbpmJobQueue"/>
                </jms-queue>
                <jms-queue name="DLQ">
                    <entry name="queue/DLQ"/>
                    <entry name="java:jboss/exported/jms/queue/DLQ"/>
                </jms-queue>
                <jms-queue name="ExpiryQueue">
                    <entry name="queue/ExpiryQueue"/>
                    <entry name="java:jboss/exported/jms/queue/ExpiryQueue"/>
                </jms-queue>
                <jms-queue name="periodicDerivationQueue">
                    <entry name="queue/bpm/periodicDerivationQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/periodicDerivationQueue"/>
                </jms-queue>
                <jms-queue name="cacheUpdateReplicationQueue">
                    <entry name="queue/bpm/cacheUpdateReplicationQueue"/>
                    <entry name="java:jboss/exported/jms/queue/bpm/cacheUpdateReplicationQueue"/>
                </jms-queue>
                <jms-topic name="asyncActionServiceStatusTopic">
                    <entry name="topic/bpm/asyncActionServiceStatusTopic"/>
                    <entry name="java:jboss/exported/jms/topic/bpm/asyncActionServiceStatusTopic"/>
                </jms-topic>
                <jms-topic name="asyncActionServiceStatusRequestTopic">
                    <entry name="topic/bpm/asyncActionServiceStatusRequestTopic"/>
                    <entry name="java:jboss/exported/jms/topic/bpm/asyncActionServiceStatusRequestTopic"/>
                </jms-topic>
                <jms-topic name="asyncActionAffinityRequestTopic">
                    <entry name="topic/bpm/asyncActionAffinityRequestTopic"/>
                    <entry name="java:jboss/exported/jms/topic/bpm/asyncActionAffinityRequestTopic"/>
                </jms-topic>
                <jms-topic name="asyncActionAffinityReplyTopic">
                    <entry name="topic/bpm/asyncActionAffinityReplyTopic"/>
                    <entry name="java:jboss/exported/jms/topic/bpm/asyncActionAffinityReplyTopic"/>
                </jms-topic>
                <jms-topic name="processEventTopic">
                    <entry name="topic/bpm/processEventTopic"/>
                    <entry name="java:jboss/exported/jms/topic/bpm/processEventTopic"/>
                </jms-topic>
                <jms-topic name="cacheUpdateReplicationEventTopic">
                    <entry name="topic/bpm/cacheUpdateReplicationEventTopic"/>
                    <entry name="java:jboss/exported/jms/topic/bpm/cacheUpdateReplicationEventTopic"/>
                </jms-topic>
            </jms-destinations>

Answer 1

ObjectMessage序列化安全性是问题所在。

ObjectMessage对象依赖于Java序列化的封送/解封送对象有效负载。此过程通常被认为是不安全的，因为恶意有效载荷可以利用主机系统。这就是为什么从5.12.2和5.13.0版本开始，ActiveMQ强制用户显式地使用ObjectMessages交换白名单包。

几天前我看到了这个，并增加了一个白名单，但它没有解决这个问题。我还试着与AMQ5.11.3竞争，但没有奏效。显然，他们也在5.11.3中增加了安全功能。无论如何，我将这个(-Dorg.apache.activemq.SERIALIZABLE_PACKAGES="*")添加到客户端和AMQ参数中，现在一切都正常工作。

请记住，我使用的命令行选项是我在代理中显式打开的安全漏洞，它允许恶意用户在我的系统上执行代码。使用该标志的正确方法是显式列出允许反序列化的类，或者最多使用包通配符来避免显式列出受信任父包中的单个类和子包。