java - 超时后重新发送消息

Question

我有一个放入 Spring AMQP 中的对象列表。对象来自 Controller 。有一个服务可以处理这些对象。并且此服务可能会因 OutOfMemoryException 而崩溃。因此，我运行了该应用程序的多个实例。

有一个问题:当服务崩溃时，我会丢失收到的消息。我读到了有关 NACK 的内容。并且可以在异常或运行时异常的情况下使用它。但我的服务因错误而崩溃。因此，我无法发送 NACK。是否可以在AMQP中设置一个超时，之后如果我没有确认之前到达的消息，我会再次发送消息？

这是我编写的代码:

public class Exchanges {
    public static final String EXC_RENDER_NAME = "render.exchange.topic";
    public static final TopicExchange EXC_RENDER = new TopicExchange(EXC_RENDER_NAME, true, false);
}

public class Queues {
    public static final String RENDER_NAME = "render.queue.topic";
    public static final Queue RENDER = new Queue(RENDER_NAME);
}

@RequiredArgsConstructor
@Service
public class RenderRabbitEventListener extends RabbitEventListener {
    private final ApplicationEventPublisher eventPublisher;

    @RabbitListener(bindings = @QueueBinding(value = @Queue(Queues.RENDER_NAME),
                                             exchange = @Exchange(value = Exchanges.EXC_RENDER_NAME, type = "topic"),
                                             key = "render.#")
    )
    public void onMessage(Message message, Channel channel) {
        String routingKey = parseRoutingKey(message);
        log.debug(String.format("Event %s", routingKey));
        RenderQueueObject queueObject = parseRender(message, RenderQueueObject.class);
        handleMessage(queueObject);
    }
    public void handleMessage(RenderQueueObject render) {
        GenericSpringEvent<RenderQueueObject> springEvent = new GenericSpringEvent<>(render);
        springEvent.setRender(true);
        eventPublisher.publishEvent(springEvent);
    }
}

这是发送消息的方法:

    @Async ("threadPoolTaskExecutor")
    @EventListener (condition = "# event.queue")
    public void start (GenericSpringEvent <RenderQueueObject> event) {
        RenderQueueObject renderQueueObject = event.getWhat ();
        send (RENDER_NAME, renderQueueObject);
}
private void send(String routingKey, Object queue) {
    try {
        rabbitTemplate.convertAndSend(routingKey, objectMapper.writeValueAsString(queue));
    } catch (JsonProcessingException e) {
        log.warn("Can't send event!", e);
    }
}

Answer 1

您需要关闭连接才能使消息重新排队。

最好在 OOME 后终止应用程序(这当然会关闭连接)。