弹簧批量和弹簧集成(JMS)和负载平衡从系统

我在用

1. 弹簧批次

   • 步骤1
   • 步骤2主(分区程序)
   • 步骤3

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

我们看到的问题是,第一个slave处理所有JMS消息,而不是在slave之间均匀分布。

配置如下

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

如果你有经验,请提出建议。

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

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