我在我的 spring 批处理项目编写器中使用 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 清除队列？