这就是我现有系统的工作方式。
我使用spring批处理编写了批处理，它将消息异步写入队列。编写者一旦将一定数量的消息发送到队列，就会开始监听 LINKED_BLOCKING_QUEUE 以获得相同数量的消息。
我有使用消息并处理它们的 spring amqp 侦听器。一旦处理完毕，消费者就会在回复队列中回复。有侦听器侦听回复队列以检查消息是否成功处理。回复侦听器检索响应并将其添加到 LINKED_BLOCKING_QUEUE，然后由 writer 获取。一旦作者获取所有响应完成批处理。如果有异常，它将停止批处理。
这是我的工作配置
<beans:bean id="computeListener" class="com.st.symfony.Foundation"
p:symfony-ref="symfony" p:replyTimeout="${compute.reply.timeout}" />
<rabbit:queue name="${compute.queue}" />
<rabbit:queue name="${compute.reply.queue}" />
<rabbit:direct-exchange name="${compute.exchange}">
<rabbit:bindings>
<rabbit:binding queue="${compute.queue}" key="${compute.routing.key}" />
</rabbit:bindings>
</rabbit:direct-exchange>
<rabbit:listener-container
connection-factory="rabbitConnectionFactory" concurrency="${compute.listener.concurrency}"
requeue-rejected="false" prefetch="1">
<rabbit:listener queues="${compute.queue}" ref="computeListener"
method="run" />
</rabbit:listener-container>
<beans:beans profile="master">
<beans:bean id="computeLbq" class="java.util.concurrent.LinkedBlockingQueue" />
<beans:bean id="computeReplyHandler" p:blockingQueue-ref="computeLbq"
class="com.st.batch.foundation.ReplyHandler" />
<rabbit:listener-container
connection-factory="rabbitConnectionFactory" concurrency="1"
requeue-rejected="false">
<rabbit:listener queues="${compute.reply.queue}" ref="computeReplyHandler"
method="onMessage" />
</rabbit:listener-container>
<beans:bean id="computeItemWriter"
class="com.st.batch.foundation.AmqpAsynchItemWriter"
p:template-ref="amqpTemplate" p:queue="${compute.queue}"
p:replyQueue="${compute.reply.queue}" p:exchange="${compute.exchange}"
p:replyTimeout="${compute.reply.timeout}" p:routingKey="${compute.routing.key}"
p:blockingQueue-ref="computeLbq"
p:logFilePath="${spring.tmp.batch.dir}/#{jobParameters[batch_id]}/log.txt"
p:admin-ref="rabbitmqAdmin" scope="step" />
<job id="computeJob" restartable="true">
<step id="computeStep">
<tasklet transaction-manager="transactionManager">
<chunk reader="computeFileItemReader" processor="computeItemProcessor"
writer="computeItemWriter" commit-interval="${compute.commit.interval}" />
</tasklet>
</step>
</job>
</beans:beans>
这是我的作家代码，
public class AmqpAsynchRpcItemWriter<T> implements ItemWriter<T> {
protected String exchange;
protected String routingKey;
protected String queue;
protected String replyQueue;
protected RabbitTemplate template;
protected AmqpAdmin admin;
BlockingQueue<Object> blockingQueue;
String logFilePath;
long replyTimeout;
// Getters and Setters
@Override
public void write(List<? extends T> items) throws Exception {
for (T item : items) {
Message message = MessageBuilder
.withBody(item.toString().getBytes())
.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
.setReplyTo(this.replyQueue)
.setCorrelationId(item.toString().getBytes()).build();
template.send(this.exchange, this.routingKey, message);
}
for (T item : items) {
Object msg = blockingQueue
.poll(this.replyTimeout, TimeUnit.MILLISECONDS);
if (msg instanceof Exception) {
admin.purgeQueue(this.queue, true);
throw (Exception) msg;
} else if (msg == null) {
throw new Exception("reply timeout...");
}
}
System.out.println("All items are processed.. Command completed. ");
}
}
听众pojo
public class Foundation {
Symfony symfony;
long replyTimeout;
//Getters Setters
public Object run(String command) {
System.out.println("Running:" + command);
try {
symfony.run(command, this.replyTimeout);
} catch (Exception e) {
return e;
}
return "Completed : " + command;
}
}
这是回复处理程序
public class ReplyHandler {
BlockingQueue<Object> blockingQueue;
public void onMessage(Object msgContent) {
try {
blockingQueue.put(msgContent);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
现在，问题是，我想同时运行具有唯一批次 ID 的多个批次，这将为不同批次处理不同的数据（相同类型）。
由于将来批次的数量会增加，我不想继续为每个批次添加单独的队列和回复队列。
而且，为了同时处理消息，我有多个侦听器（使用侦听器并发设置）侦听队列。如果我为不同的批次添加不同的队列，运行的监听器数量将会增加，这可能会使服务器过载（CPU/内存使用率很高）。
因此，我不想为要添加的每种类型的批次复制相同的基础架构。我想使用相同的基础设施，只是特定批次的编写者应该只得到它的响应，而不是同时运行的其他批次的响应。
我们可以使用相同的项目编写器实例，这些实例对并行运行的多个批处理实例使用相同的阻塞队列实例吗？