java - SimpleMessageListenerContainer - 从失败中恢复

Question

在任何可能的错误(缺少队列、连接问题等)上关闭 SimpleMessageListenerContainer(以编程方式创建，而不是作为 bean)并创建新的(在运行时重新声明所有绑定(bind))的最简单方法是什么。

我使用 helix 进行分区管理，每个分区有 1 个监听器。一种可能性是也使用现有的 SimpleMessageListenerContainer(并不总是创建新的)，但在这种情况下，我需要重试队列重新声明和重新绑定(bind)，以防出现任何失败。

此外，似乎还有不同类型的异常 - 致命的(例如，在运行时删除队列)和非致命的(连接丢失)。如何同时处理这两种情况？

这两个选项哪个更容易？

已更新

private Map<SimpleMessageListenerContainer, AtomicBoolean> shuttingDown = new ConcurrentHashMap<>();

@Override
public void onApplicationEvent(ListenerContainerConsumerFailedEvent listenerContainerConsumerFailedEvent) {

    boolean fatal = listenerContainerConsumerFailedEvent.isFatal();
    SimpleMessageListenerContainer listenerContainer = (SimpleMessageListenerContainer)listenerContainerConsumerFailedEvent.getSource();

    if(fatal){
        AtomicBoolean sd = shuttingDown
                .computeIfAbsent(listenerContainer, v -> new AtomicBoolean(false));
        if(sd.compareAndSet(false, true)) {
            System.out.println("RECREATING");
            String[] qn = listenerContainer.getQueueNames();
            String q = qn[0];
            recreateQueue(q);
            listenerContainer.stop();
            listenerContainer.start();
            //delete from shuttingDown ?
        }
        else{
            System.out.println("RECREATING_NOT");
        }
    }
    else{
        System.out.println("NON_FATAL");
    }
}

和输出

NON_FATAL
NON_FATAL
NON_FATAL
NON_FATAL
22:36:44.044 [SimpleAsyncTaskExecutor-7] ERROR org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer - Consumer received fatal=false...\
...

RECREATING
RECREATING_NOT
RECREATING_NOT
RECREATING_NOT
22:36:44.057 [SimpleAsyncTaskExecutor-6] ERROR org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer - Stopping container from aborted consumer

Answer 1

向容器添加一个ApplicationEventPublisher； ListenerContainerConsumerFailedEvent有一个fatal boolean 属性。

编辑

@SpringBootApplication
public class So47357940Application {

    public static void main(String[] args) {
        SpringApplication.run(So47357940Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(AmqpAdmin admin) {
        return args -> admin.deleteQueue("so47357940");
    }

    @RabbitListener(queues = "so47357940")
    public void listen(String in) {
        System.out.println(in);
    }

    private final Map<SimpleMessageListenerContainer, AtomicBoolean> shuttingDown = new ConcurrentHashMap<>();

    @Bean
    public ApplicationListener<ListenerContainerConsumerFailedEvent> failures(AmqpAdmin admin,
            RabbitTemplate template) {
        return event -> {
            if (event.isFatal()) {
                SimpleMessageListenerContainer container = (SimpleMessageListenerContainer) event.getSource();
                AtomicBoolean sd = this.shuttingDown.computeIfAbsent(container, v -> new AtomicBoolean());
                if (sd.compareAndSet(false, true)) {
                    System.out.println("RECREATING");
                    String[] qn = container.getQueueNames();
                    String q = qn[0];
                    admin.declareQueue(new Queue(q));
                    // better to use a shared exec
                    ExecutorService exec = Executors.newSingleThreadExecutor();
                    exec.execute(() -> {
                        while (container.isRunning()) {
                            // should probably give up at some point
                            try {
                                Thread.sleep(100);
                            }
                            catch (InterruptedException e) {
                                Thread.currentThread().interrupt();
                            }
                        }
                        container.start();
                        template.convertAndSend("so47357940", "foo");
                        this.shuttingDown.remove(container);
                    });
                }
                else {
                    System.out.println("RECREATING_NOT");
                }
            }
            else {
                System.out.println("NON_FATAL");
            }
        };
    }

}

这是我得到的调试日志...

RECREATING
2017-11-17 17:38:53.893 DEBUG 42372 --- [cTaskExecutor-2] o.s.a.r.c.CachingConnectionFactory       : Creating cached Rabbit Channel from AMQChannel(amqp://guest@127.0.0.1:5672/,3)
2017-11-17 17:38:53.893 DEBUG 42372 --- [cTaskExecutor-2] o.s.amqp.rabbit.core.RabbitTemplate      : Executing callback RabbitAdmin$$Lambda$144/1094003461 on RabbitMQ Channel: Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473]
2017-11-17 17:38:53.893 DEBUG 42372 --- [cTaskExecutor-2] o.s.amqp.rabbit.core.RabbitAdmin         : declaring Queue 'so47357940'
2017-11-17 17:38:53.901 DEBUG 42372 --- [cTaskExecutor-2] o.s.a.r.l.SimpleMessageListenerContainer : Cancelling Consumer@3a813488: tags=[Cancelling Consumer@3a813488: tags=[{}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,2), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473], acknowledgeMode=AUTO local queue size=0], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,2), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473], acknowledgeMode=AUTO local queue size=0
2017-11-17 17:38:53.901 DEBUG 42372 --- [cTaskExecutor-2] o.s.a.r.listener.BlockingQueueConsumer   : Closing Rabbit Channel: Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,2), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473]
2017-11-17 17:38:53.901 DEBUG 42372 --- [cTaskExecutor-2] o.s.a.r.c.CachingConnectionFactory       : Closing cached Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,2)
2017-11-17 17:38:53.903 ERROR 42372 --- [cTaskExecutor-2] o.s.a.r.l.SimpleMessageListenerContainer : Stopping container from aborted consumer
2017-11-17 17:38:53.903 DEBUG 42372 --- [cTaskExecutor-2] o.s.a.r.l.SimpleMessageListenerContainer : Shutting down Rabbit listener container
2017-11-17 17:38:53.903  INFO 42372 --- [cTaskExecutor-2] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
2017-11-17 17:38:53.903  INFO 42372 --- [cTaskExecutor-2] o.s.a.r.l.SimpleMessageListenerContainer : Successfully waited for workers to finish.
2017-11-17 17:38:54.003 DEBUG 42372 --- [pool-4-thread-1] o.s.a.r.l.SimpleMessageListenerContainer : Starting Rabbit listener container.
2017-11-17 17:38:54.004 DEBUG 42372 --- [cTaskExecutor-3] o.s.a.r.listener.BlockingQueueConsumer   : Starting consumer Consumer@3a2547b8: tags=[Starting consumer Consumer@3a2547b8: tags=[{}], channel=null, acknowledgeMode=AUTO local queue size=0], channel=null, acknowledgeMode=AUTO local queue size=0
2017-11-17 17:38:54.005 DEBUG 42372 --- [cTaskExecutor-3] o.s.a.r.listener.BlockingQueueConsumer   : Started on queue 'so47357940' with tag amq.ctag-3wMG_13-68ibLL05ir3ySA: Consumer@3a2547b8: tags=[{amq.ctag-3wMG_13-68ibLL05ir3ySA=so47357940}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473], acknowledgeMode=AUTO local queue size=0
2017-11-17 17:38:54.005 DEBUG 42372 --- [ool-1-thread-11] o.s.a.r.listener.BlockingQueueConsumer   : ConsumeOK : Consumer@3a2547b8: tags=[{amq.ctag-3wMG_13-68ibLL05ir3ySA=so47357940}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473], acknowledgeMode=AUTO local queue size=0
2017-11-17 17:38:54.008 DEBUG 42372 --- [pool-4-thread-1] o.s.a.r.c.CachingConnectionFactory       : Creating cached Rabbit Channel from AMQChannel(amqp://guest@127.0.0.1:5672/,4)
2017-11-17 17:38:54.008 DEBUG 42372 --- [pool-4-thread-1] o.s.amqp.rabbit.core.RabbitTemplate      : Executing callback RabbitTemplate$$Lambda$146/1108520685 on RabbitMQ Channel: Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,4), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473]
2017-11-17 17:38:54.008 DEBUG 42372 --- [pool-4-thread-1] o.s.amqp.rabbit.core.RabbitTemplate      : Publishing message (Body:'foo' MessageProperties [headers=Publishing message (Body:'foo' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=3, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])on exchange [], routingKey = [so47357940], contentType=text/plain, contentEncoding=UTF-8, contentLength=3, deliveryMode=PERSISTENT, priority=0, deliveryTag=0])on exchange [], routingKey = [so47357940]
2017-11-17 17:38:54.012 DEBUG 42372 --- [ool-1-thread-12] o.s.a.r.listener.BlockingQueueConsumer   : Storing delivery for Consumer@3a2547b8: tags=[{amq.ctag-3wMG_13-68ibLL05ir3ySA=so47357940}], channel=Cached Rabbit Channel: AMQChannel(amqp://guest@127.0.0.1:5672/,3), conn: Proxy@48c9a496 Shared Rabbit Connection: SimpleConnection@3bdd3f4a [delegate=amqp://guest@127.0.0.1:5672/, localPort= 55473], acknowledgeMode=AUTO local queue size=0
2017-11-17 17:38:54.012 DEBUG 42372 --- [cTaskExecutor-3] o.s.a.r.listener.BlockingQueueConsumer   : Received message: (Body:'foo' MessageProperties [headers=Received message: (Body:'foo' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=so47357940, deliveryTag=1, consumerTag=amq.ctag-3wMG_13-68ibLL05ir3ySA, consumerQueue=so47357940]), contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=so47357940, deliveryTag=1, consumerTag=amq.ctag-3wMG_13-68ibLL05ir3ySA, consumerQueue=so47357940])
2017-11-17 17:38:54.015 DEBUG 42372 --- [cTaskExecutor-3] .a.r.l.a.MessagingMessageListenerAdapter : Processing [GenericMessage [payload=foo, headers={amqp_receivedDeliveryMode=PERSISTENT, amqp_receivedRoutingKey=so47357940, amqp_contentEncoding=UTF-8, amqp_deliveryTag=1, amqp_consumerQueue=so47357940, amqp_redelivered=false, id=b614d9e6-1744-b600-7d86-ca9c51ad5844, amqp_consumerTag=amq.ctag-3wMG_13-68ibLL05ir3ySA, contentType=text/plain, timestamp=1510958334014}]]
foo