使用spring amqp模板清除rabbitmq队列？

Question

我在spring spring item writer中使用spring amqp模板向rabbitmq队列添加消息.

public class AmqpAsynchRpcItemWriter<T> implements ItemWriter<T> {

    protected String exchange;
    protected String routingKey;
    protected String queue;
    protected String replyQueue;
    protected RabbitTemplate template;
    BlockingQueue<Object> blockingQueue;


    public void onMessage(Object msgContent) {

        try {
            blockingQueue.put(msgContent);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

    @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(60, TimeUnit.SECONDS);

            if (msg instanceof Exception) {
                throw (Exception) msg;
            } else if (msg == null) {
                System.out.println("reply timeout...");
                break;
            } 

        }
    }

}

消息将在不同的远程服务器上处理.我正在尝试处理用例,如果我的消息处理失败(由于某些异常),则将停止执行步骤.

我想清除该队列中的所有剩余消息,以便队列中剩余的消息不应被消耗和处理,因为它们也将失败.

如果步骤失败,我的项目编写器将再次排队所有消息,因此我需要清除任何异常上的所有剩余消息.

如何使用spring amqp清除队列？

Answer 1

我可以用它做

admin.purgeQueue(this.queue,true);