如何同时运行共享相同读取器和写入器实例的spring批处理作业？

Question

这就是我现有系统的工作方式。

我使用spring批处理编写了批处理，它将消息异步写入队列。Writer一旦向queue发送了一定数量的消息，就开始监听LINKED_BLOCKING_QUEUE中相同数量的消息。

我有spring amqp监听器，它使用并处理消息。一旦处理完毕，使用者就会在回复队列中回复。有侦听器侦听应答队列，以检查消息是否被成功处理。应答侦听器检索响应并将其添加到LINKED_BLOCKING_QUEUE，然后由Writer获取。一旦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>

这是我的writer代码,

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();

        }
    }

}

对于并行运行的多个批处理实例，我们是否可以使用相同的项编写器实例，这些项编写器使用相同的阻塞队列实例？

Answer 1

您可能需要查看JMS消息选择器。

根据单据

createConsumer和createDurableSubscriber方法允许您在创建消息使用者时将消息选择器指定为参数。