rabbitmq Consumer 丢失

Question

问题遇到的现象和发生背景

如上图，因为业务原因导致 消费者丢失，错误信息如下，目前该错误已经找到并解决，
o.s.a.r.l.SimpleMessageListenerContainer - Consumer thread error, thread abort. java.lang.AssertionError: Failed generating bytecode for :15
但是后面可能会出现其他场景导致Consumers 丢失。所以想请教下大家，有没有遇到过类似的问题，都如何处理的，

我想要达到的结果

rabbitMq 是否怎么设置 能让消费者重新生成并连接。

Answer 1

1.设置消费者自动重连
在RabbitMQ客户端中，可以通过设置 requeueRejected 或 republishRejected 属性来控制消费者在出现错误或异常时是否自动重新连接。具体实现如下：

SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
container.setConnectionFactory(connectionFactory);
container.setQueueNames(queueName);
container.setMessageListener(messageListenerAdapter);

// 设置消费者自动重连
container.setRequeueRejected(false); // 当出现异常时，将消息重新放回队列

2.监控消费者状态并重新生成
另一种处理方式是监控消费者状态，并在消费者异常或错误时重新生成新的消费者。可以使用Spring Boot Actuator或其他监控工具监控消费者状态，并在出现异常或错误时重新生成消费者。以下是一些可供参考的代码：

// 监控消费者状态
@Autowired
RabbitListenerEndpointRegistry endpointRegistry;

@Bean
public EndpointHealthIndicator rabbitHealthIndicator() {
    return new RabbitHealthIndicator(endpointRegistry);
}

// 在出现异常或错误时重新生成消费者
@RabbitListener(queues = "myQueue")
public void onMessage(Message message, Channel channel) throws IOException {
    try {
        // 处理消息
    } catch (Exception ex) {
        // 重新生成消费者
        endpointRegistry.getListenerContainers().stream()
            .filter(c -> c.isRunning() && c.getQueueNames().contains("myQueue"))
            .forEach(c -> {
                c.stop();
                c.start();
            });
    }
}