Spring Batch 和 Spring Integration (JMS) 和负载平衡从站

Question

我正在使用

1. 春季批次

   • 步骤 1
   • 步骤 2 主控（分区）
   • 第三步

2. Spring 集成 (JMS) 用于主从通信

我们看到的问题是，第一个从站处理所有 JMS 消息，而不是在从站之间平均分配。

请看下面的配置

1. 大师

   <bean id="PreProcess" class="com.job.tasklet.PreProcessTasklet" scope="step">
       <constructor-arg index="0" value="${run.slave}"/>
       <property name="maxNumberOfSlaves" value="#{jobParameters['max-slave-count']}"/>
   </bean>
   
   <bean id="PostProcess" class="com.job.tasklet.PostProcessTasklet" scope="prototype">
       <constructor-arg index="0" ref="chpsJobDataSource"/>
   </bean>
   
   
   <bean id="partitioner" class="com.job.partition.DatabasePartitioner" scope="step">
       <constructor-arg index="3" value="${max.row.count}"/>
   </bean>
   
   <bean id="partitionHandler" class="com.job.handler.StepExecutionAggregatorHandler">
       <property name="stepName" value="processAutoHoldSlaveStep"/>
       <property name="gridSize" value="${grid.size}"/>
       <property name="replyChannel" ref="aggregatedGroupRuleReplyChannel"/>
       <property name="messagingOperations">
           <bean class="org.springframework.integration.core.MessagingTemplate">
               <property name="defaultChannel" ref="groupRuleRequestsChannel"/>
           </bean>
       </property>
   </bean>
   

   <!-- Request Start -->
   <int:channel id="groupRuleRequestsChannel" />
   <int-jms:outbound-channel-adapter channel="groupRuleRequestsChannel" jms-template="jmsTemplateToSlave"/>
   
   <bean id="jmsTemplateToSlave" class="org.springframework.jms.core.JmsTemplate">
       <property name="connectionFactory" ref="connectionFactory"/>
       <property name="receiveTimeout" value="5000"/>
       <property name="defaultDestinationName" value="defaultRequest"/>
   </bean>
   
   <bean id="jmsTemplateFromSlave" class="org.springframework.jms.core.JmsTemplate" parent="jmsTemplateToSlave">
       <property name="defaultDestinationName" value="defaultRequest"/>
   </bean>
   
   
   <!-- Response Test Start -->
   <int:channel id="groupRuleReplyChannel">
       <!-- <int:queue/> -->
   </int:channel>
   
   <int-jms:inbound-channel-adapter channel="groupRuleReplyChannel" jms-template="jmsTemplateFromSlave">
       <int:poller id="defaultPoller" default="true" max-messages-per-poll="1" fixed-rate="3000"  />
   </int-jms:inbound-channel-adapter>
   
   <!-- define aggregatedReplyChannel -->
   <int:channel id="aggregatedGroupRuleReplyChannel">
       <int:queue/>
   </int:channel>
   
   <int:aggregator ref="partitionHandler"
                   input-channel="groupRuleReplyChannel"
                   output-channel="aggregatedGroupRuleReplyChannel"
                   send-timeout="3600000"/>
   

2. 从属

   <int:channel id="requestsChannel" />
   
   <bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory">
       <property name="brokerURL" value="${spring.activemq.broker-url}" />
       <property name="trustAllPackages" value="true" />
   </bean>
   
    <int-jms:message-driven-channel-adapter id="jmsIn" destination-name="#{args[0]}" channel="requestsChannel" connection-factory="connectionFactory" max-messages-per-task="1"/>
   
   <int:service-activator input-channel="requestsChannel" output-channel="replyChannel" ref="stepExecutionRequestHandler" />
   
   <int:channel id="replyChannel" />
   
   <int-jms:outbound-channel-adapter connection-factory="connectionFactory" destination-name="#{args[1]}" channel="replyChannel" />
   

如果您遇到此问题，请提出建议。

如果您需要更多信息，请告诉我。

注意：我已经在这里和谷歌搜索了很多，但还没有找到解决方案。

Answer 1

ActiveMQ 默认使用预取 1000 see here。

换句话说，前（最多）1000 个分区将分配给第一个消费者等。

可以减少预取； 1 可能适用于此应用程序。