【发布时间】：2014-07-29 00:17:39
【问题描述】：
我已经使用 spring 批处理分区并使用 rabbitmq 作为中间件来实现我的批处理作业。
我研究了文档并参考了这些单元测试
https://github.com/sshcherbakov/spring-batch-talk/blob/master/src/main/resources/META-INF/master.xml
https://github.com/sshcherbakov/spring-batch-talk/blob/master/src/main/resources/META-INF/slave.xml
我可以同时运行我的作业步骤，但我有点担心如果我同时使用不同参数启动同一作业的多个实例，它将如何工作。
例如，我正在使用 importExchange 作业导入交易所数据，但如果我同时为不同市场（例如美国市场、欧洲市场）启动 importExchange 作业。
Partitioner 将输入交换名称分区到不同的分区步骤执行上下文中，MessagePartitionHandler 将 stepExecutionRequests 作为消息通过 rabbitmq 队列发送到不同的服务器，并将在不同的服务器上同时执行步骤。
现在令人困惑的是，当响应被发送回回复队列（所有作业实例都相同）时，所有实例的侦听器将侦听相同的回复队列。例如 job1 和 job2 回复都将在同一个回复队列上监听。
我们如何确保 job1 的响应将被 job1 出站网关而不是由 job2 接收，反之亦然？出站网关是否只接收对自身发送的请求的响应（通过检查关联 ID）而忽略其他响应？
由于我们使用的是直接通道和交换，响应将只传递给一个侦听器，所以作业1的响应会被作业2的侦听器拾取吗？
或者是否有任何路由器或过滤器可以选择性地选择回复？
我需要担心这个还是 MessageChannelPartitionHandler 负责这个？或者我应该在回复队列前面加上作业 ID 吗？
下面是我的配置
<task:executor id="taskExecutor" pool-size="20" />
<int:channel id="importExchangesOutboundChannel">
<int:dispatcher task-executor="taskExecutor" />
</int:channel>
<int:channel id="importExchangesInboundStagingChannel" />
<amqp:outbound-gateway request-channel="importExchangesOutboundChannel"
reply-channel="importExchangesInboundStagingChannel" amqp-template="importExchangesAmqpTemplate"
mapped-request-headers="correlationId, sequenceNumber, sequenceSize, STANDARD_REQUEST_HEADERS"
mapped-reply-headers="correlationId, sequenceNumber, sequenceSize, STANDARD_REQUEST_HEADERS" />
<beans:bean id="importExchangesMessagingTemplate"
class="org.springframework.integration.core.MessagingTemplate"
p:defaultChannel-ref="importExchangesOutboundChannel"
p:receiveTimeout="150000" />
<beans:bean id="importExchangesPartitioner"
class="org.springframework.batch.core.partition.support.FlatFilePartitioner"
p:resource="file:${spring.tmp.batch.dir}/#{jobParameters[batch_id]}/exchanges.txt"
scope="step" />
<beans:bean id="importExchangesPartitionHandler"
class="org.springframework.batch.integration.partition.MessageChannelPartitionHandler"
p:stepName="importExchangesStep" p:gridSize="6"
p:messagingOperations-ref="importExchangesMessagingTemplate" />
<int:aggregator ref="importExchangesPartitionHandler"
send-partial-result-on-expiry="true" send-timeout="300000"
input-channel="importExchangesInboundStagingChannel" />
<amqp:inbound-gateway concurrent-consumers="6"
request-channel="importExchangesInboundChannel" receive-timeout="300000"
reply-channel="importExchangesOutboundStagingChannel" queue-names="importExchangesQueue"
connection-factory="rabbitConnectionFactory"
mapped-request-headers="correlationId, sequenceNumber, sequenceSize, STANDARD_REQUEST_HEADERS"
mapped-reply-headers="correlationId, sequenceNumber, sequenceSize, STANDARD_REQUEST_HEADERS" />
<rabbit:template id="importExchangesAmqpTemplate" connection-factory="rabbitConnectionFactory"
routing-key="importExchangesQueue" reply-timeout="300000">
</rabbit:template>
<int:channel id="importExchangesInboundChannel" />
<int:service-activator ref="stepExecutionRequestHandler"
input-channel="importExchangesInboundChannel" output-channel="importExchangesOutboundStagingChannel" />
<int:channel id="importExchangesOutboundStagingChannel" />
<rabbit:queue name="${import.exchanges.queue}" />
<rabbit:queue name="${import.exchanges.reply.queue}" />
<rabbit:direct-exchange name="${import.exchanges.exchange}">
<rabbit:bindings>
<rabbit:binding queue="${import.exchanges.queue}"
key="${import.exchanges.routing.key}" />
</rabbit:bindings>
</rabbit:direct-exchange>
<beans:bean id="stepExecutionRequestHandler"
class="org.springframework.batch.integration.partition.StepExecutionRequestHandler"
p:jobExplorer-ref="jobExplorer" p:stepLocator-ref="stepLocator" />
<beans:bean id="stepLocator"
class="org.springframework.batch.integration.partition.BeanFactoryStepLocator" />
<beans:bean id="importExchangesItemWriter"
class="com.st.batch.foundation.ImportExchangesItemWriter"
p:symfony-ref="symfony" p:replyTimeout="${import.exchanges.reply.timeout}"
p:logFilePath="${batch.log.file.path}.#{jobParameters[batch_id]}"
scope="step" />
<beans:bean id="importExchangesFileItemReader"
class="org.springframework.batch.item.file.MultiThreadedFlatFileItemReader"
p:resource="file:${spring.tmp.batch.dir}/#{jobParameters[batch_id]}/exchanges.txt"
p:lineMapper-ref="stLineMapper" p:startAt="#{stepExecutionContext['startAt']}"
p:maxItemCount="#{stepExecutionContext['itemsCount']}" scope="step" />
<step id="importExchangesStep">
<tasklet transaction-manager="transactionManager">
<chunk reader="importExchangesFileItemReader" writer="importExchangesItemWriter"
commit-interval="${import.exchanges.commit.interval}" />
</tasklet>
</step>
<job id="importExchangesJob" restartable="true">
<step id="importExchangesStep.master" next="importEclsStep.master">
<partition partitioner="importExchangesPartitioner"
handler="importExchangesPartitionHandler" />
</step>
</job>
编辑：
我尝试从 amqpTemplate 中删除回复队列名称以使用默认的临时回复队列并测试了这个用例，在查看回复之前，问题也出在从属端。
<rabbit:template id="importExchangesAmqpTemplate" connection-factory="rabbitConnectionFactory"
routing-key="importExchangesQueue" reply-timeout="300000">
</rabbit:template>
我创建了两个带有虚拟数据的输入文件，例如
我的工作 ID 是 2014-06-08 和 2014-06-09。我在文件夹名称 2014-06-08 和 2014-06-09 下创建了 exchange.txt。
/home/ubuntu/tmp/spring/batch/2015-06-08/exchanges.txt
/home/ubuntu/tmp/spring/batch/2015-06-09/exchanges.txt
/home/ubuntu/tmp/spring/batch/2015-06-08/exchanges.txt文件中的数据是
1
2
3
up to 30
在 /home/ubuntu/tmp/spring/batch/2015-06-09/exchanges.txt 中是
31
32
33
up to 60
我正在使用这个项目阅读器来读取项目并传递给作者。
读者：
<beans:bean id="importExchangesFileItemReader"
class="org.springframework.batch.item.file.MultiThreadedFlatFileItemReader"
p:resource="file:${spring.tmp.batch.dir}/#{jobParameters[batch_id]}/exchanges.txt"
p:lineMapper-ref="stLineMapper" p:startAt="#{stepExecutionContext['startAt']}"
p:maxItemCount="#{stepExecutionContext['itemsCount']}" scope="step" />
作者：
<beans:bean id="importExchangesItemWriter"
class="com.st.batch.foundation.ImportExchangesItemWriter"
p:symfony-ref="symfony" p:replyTimeout="${import.ecls.reply.timeout}"
p:logFilePath="${batch.log.file.path}.#{jobParameters[batch_id]}"
scope="step" />
内部作者我正在调用外部命令，该命令为每个项目交换导入数据
@Override
public void write(List<? extends T> exchanges) throws Exception {
commandRunner.setLogFilePath(this.logFilePath);
for (T exchange : exchanges) {
String command = commandRunner.getConsolePath() + " "
+ "st:import exchange" + " " + exchange.toString();
commandRunner.run(command, this.replyTimeout);
}
}
在 commandRunner 内部，
public void run(String command, long replyTimeout)
throws Exception {
String[] commands = command.split("\\s+");
ProcessBuilder pb = new ProcessBuilder(commands);
File log = new File(this.logFilePath);
pb.redirectErrorStream(true);
pb.redirectOutput(Redirect.appendTo(log));
Process p = pb.start();
.......
}
如果我只启动一个作业实例（八个批处理 ID 为 2015-06-08 或 2015-06-09），一切正常，但如果我同时启动这两个作业实例的步骤的输入数据会混合，我的意思是，这就是我在日志文件中得到的内容
tail -f /var/log/st/batch.log.2015-06-08
14 23 1 27 19 9 15 24 2 10 28 20 25 16 3 21 29 11 26 17 4 30 12 22 18 5 44 45 46
并在 /var/log/st/batch.log.2015-06-09
52 13 47 31 37 6 53 57 48 32 38 54 7 49 58 33 39 55 8 59 50 34 40 56 60 51 35 42 41 36 43
所以 44 45 46 转到 batch.log.2015-06-08 应该转到 batch.log.2015-06-09 和 6 7 8 转到 batch.log.2015-06-09 应该转到batch.log.2015-06-08
我将日志文件路径传递给项目编写器，因为我需要为每个作业单独的日志文件，因此将 batch_id 附加到文件名。
<beans:bean id="importExchangesItemWriter"
class="com.st.batch.foundation.ImportExchangesItemWriter"
p:symfony-ref="symfony" p:replyTimeout="${import.exchanges.reply.timeout}"
p:logFilePath="${batch.log.file.path}.#{jobParameters[batch_id]}"
scope="step" />
是否由于出站和入站网关而发生？是否为不同的作业实例创建了不同的 Spring 集成通道、网关等实例，或者它们就像所有作业实例都相同的 rabbitmq 队列？
入站网关有 concurrent-consumers="8" 这些消费者对于所有作业实例是否相同，或者将为每个作业实例创建单独的 8 个消费者？
这个处理程序可以为多个作业分区吗？
<beans:bean id="importExchangesPartitioner"
class="org.springframework.batch.core.partition.support.FlatFilePartitioner"
p:resource="file:${spring.tmp.batch.dir}/#{jobParameters[batch_id]}/exchanges.txt"
scope="step" />
重现此问题的应用程序在这里
https://github.com/vishalmelmatti/spring-batch-remote-partition-test/tree/master
【问题讨论】：
标签： spring spring-batch spring-integration spring-amqp