Потребительская очередь удаляется, хотя уровень кэша установлен в CACHE_CONSUMER - PullRequest
Новая
       0

Потребительская очередь удаляется, хотя уровень кэша установлен в CACHE_CONSUMER

2 голосов
/ 15 ноября 2011

Я понял, даже после явной установки свойства 

<property name="cacheLevelName" value="CACHE_CONSUMER"/>

мой потребитель все еще удален.

См. Журнал ниже: 

23:14:25,180 | Usage | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 |     Main:memory:queue://reply:memory: usage change from: 0% of available memory, to: 1% of available memory
23:14:25,180 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | reply toPageIn: 1, Inflight: 0, pagedInMessages.size 2, enqueueCount: 2, dequeueCount: 1
23:14:25,180 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | reply toPageIn: 0, Inflight: 0, pagedInMessages.size 3, enqueueCount: 2, dequeueCount: 1
23:14:25,180 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | reply toPageIn: 0, Inflight: 0, pagedInMessages.size 3, enqueueCount: 2, dequeueCount: 1
23:14:25,649 | ActiveMQMessageConsumer | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | remove: ID:TestESB-3364-1321307776420-8:1:1:84, lastDeliveredSequenceId:0
23:14:25,649 | AbstractRegion | 49 - org.apache.activemq.activemq-core -     5.5.0.fuse-00-43 | default removing consumer: ID:TestESB-3364-1321307776420-8:1:1:84 for destination: queue://routerResponse
23:14:25,649 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | queue://routerResponse remove sub: QueueSubscription: consumer=ID:TestESB-3364-1321307776420-8:1:1:84, destinations=1, dispatched=0, delivered=0, pending=0, lastDeliveredSeqId: 0, dequeues: 0, dispatched: 0, inflight: 0
23:14:25,649 | AbstractRegion | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | default adding consumer: ID:TestESB-3364-1321307776420-8:1:1:85 for destination: queue://routerResponse
23:14:25,649 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | queue://routerResponse add sub: QueueSubscription: consumer=ID:TestESB-3364-1321307776420-8:1:1:85, destinations=0, dispatched=0, delivered=0, pending=0, dequeues: 0, dispatched: 0, inflight: 0
23:14:26,165 | ActiveMQMessageConsumer | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | remove: ID:TestESB-3364-1321307776420-14:1:1:4, lastDeliveredSequenceId:0
23:14:26,165 | AbstractRegion | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | default removing consumer: ID:TestESB-3364-1321307776420-14:1:1:4 for destination: queue://reply
23:14:26,165 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | queue://reply remove sub: QueueSubscription: consumer=ID:TestESB-3364-1321307776420-14:1:1:4, destinations=1, dispatched=0, delivered=0, pending=0, lastDeliveredSeqId: 0, dequeues: 1, dispatched: 1, inflight: 0
23:14:26,165 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | reply toPageIn: 0, Inflight: 0, pagedInMessages.size 3, enqueueCount: 2, dequeueCount: 1
23:14:26,165 | AbstractRegion | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | default adding consumer: ID:TestESB-3364-1321307776420-14:1:1:5 for destination: queue://reply
23:14:26,165 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | queue://reply add sub: QueueSubscription: consumer=ID:TestESB-3364-1321307776420-14:1:1:5, destinations=0, dispatched=0, delivered=0, pending=0, dequeues: 1, dispatched: 1, inflight: 0
23:14:26,165 | Queue | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | reply toPageIn: 0, Inflight: 0, pagedInMessages.size 3, enqueueCount: 2, dequeueCount: 1
23:14:26,649 | ActiveMQMessageConsumer | 49 - org.apache.activemq.activemq-core - 5.5.0.fuse-00-43 | remove: ID:TestESB-3364-1321307776420-8:1:1:85, lastDeliveredSequenceId:0

См. Мою конфигурацию верблюжьей пружины ниже: 

<bean id="providerConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<!--<property name="brokerURL"><value>${aero-provider-broker-url}</value></property>-->


<bean id="providerActiveMQConfig" class="org.apache.activemq.camel.component.ActiveMQConfiguration">
  <property name="connectionFactory" ref="providerConnectionFactory"/>
  <property name="cacheLevelName" value="CACHE_CONSUMER"/>
  <property name="concurrentConsumers" value="${jms-concurrent-consumers}"/>
  <property name="requestTimeout" value="${jms-request-timeout}"/>
  <property name="priority" value="${jms-message-priority}"/>
</bean>

<bean id="providerActivemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
  <property name="configuration" ref="providerActiveMQConfig"/>
  <property name="brokerURL"><value>${provider-broker-url}</value></property>
</bean>

Редактировать здесь >>>>

Определение бина: 

   <bean id="localJMSConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="alwaysSessionAsync" value="false"/>
    <property name="alwaysSyncSend" value="true"/>
    <property name="brokerURL"><value>${local-broker-url}</value></property>
    <property name="closeTimeout" value="150000"/>
    <property name="copyMessageOnSend" value="true"/>
    <property name="disableTimeStampsByDefault" value="false"/>
    <property name="dispatchAsync" value="false"/>
    <property name="objectMessageSerializationDefered" value="false"/>
    <property name="optimizeAcknowledge" value="false"/>
    <property name="optimizedMessageDispatch" value="true"/>
    <property name="producerWindowSize" value="0"/>
    <property name="statsEnabled" value="false"/>
    <property name="useAsyncSend" value="false"/>
    <property name="useCompression" value="false"/>
    <property name="sendTimeout" value="0"/>
</bean> 
<bean id="localJMSTransactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
    <property name="connectionFactory" ref="localJMSConnectionFactory" />               
</bean> 
<bean id="localRequestJMSConfig" class="org.apache.camel.component.jms.JmsConfiguration">
        <property name="connectionFactory" ref="localJMSConnectionFactory"/>
    <property name="concurrentConsumers" value="${jms-concurrent-consumers}"/>
    <property name="deliveryPersistent" value="true"/>
    <property name="priority" value="9"/>
</bean> 
<bean id="localRequestJMS" class="org.apache.camel.component.jms.JmsComponent">
    <property name="configuration" ref="localRequestJMSConfig"/>        
</bean>

Определение верблюжьей точки: 

<camel:endpoint id="jmsEndpoint" camelContextId="bohProcessorCC" uri="localRequestJMS:queue:${boh-processor-queue}"/>

<camel:endpoint id="providerEndpoint" camelContextId="bohProcessorCC" uri="providerActivemq:queue:${generic-http-provider-queue}?exchangePattern=InOut&amp;requestTimeout=${jms-request-timeout}&amp;replyTo=reply"/>

Верблюжий путь: 

 <from ref="jmsEndpoint"/>
 <bean ref="service" method="getSignInRequest"/> 
 <transform><simple>XML=${in.body}</simple></transform> 
 <inOut ref="providerEndpoint"/> 
 <bean ref="service" method="getSignInResponseData"/>  
 <bean ref="utilityServicesBean" method="process"/> 
 <bean ref="service" method="getPassword"/> 
 <inOut ref="providerEndpoint"/> 
 <bean ref="camelPropertyEnricherBean" method="addCustReference"/> 
 <bean ref="camelPropertyEnricherBean" method="normalizeXML"/> 
 <inOut ref="providerEndpoint"/>

Во время второго разговора время ожидания обмена обычно составляет время ожидания. Обычно в очереди ответов есть ответ, но в журнале говорится, что идентификатор корреляции не совпадает. Я заметил, что если я прошу верблюда установить идентификатор сообщения в качестве идентификатора корреляции непосредственно перед вызовом конечной точки поставщика, он обычно возвращается с сообщением. Даже это иногда дает сбой при большом трафике.

1 Ответ

2 голосов
/ 16 ноября 2011

Вы используете запрос / ответ через JMS с верблюдом, например, вы делаете inOut ref = "providerEndpoint", который также использует фиксированное имя очереди replyTo, так как вы указываете replyTo = reply.

Это означает, что Camel заставляет использовать CACHE_SESSION, поскольку очередь ответов считается общей очередью ответов, а Camel использует селектор сообщений JMS для получения ожидаемого ответного сообщения на основе идентификаторов корреляции. А для обновления селектора сообщений JMS необходимо заново создать получателя. И именно поэтому вы видите, что потребитель удален в журналах. Это тот ответный потребитель.

В Camel 2.9 и далее мы ввели опцию replyToType, которую вы можете установить в replyToType = Exclusive. И тогда очередь ответа считается исключительной, что означает, что Camel вместо этого использует CACHE_CONSUMER.

...