스프링 래빗엠큐의 셧다운 시간을 변경하는 방법이다.

스프링 래빗엠큐의 셧다운 기본 설정

스프링 애플리케이션의 종료

스프링 애플리케이션은 종료 시그널을 받고 셧다운을 시작한다.

로그 샘플은 아래와 같고,

로그에서는 래빗엠큐(SimpleBrokerMessageHandler, SimpleMessageListenerContainer), 서블릿 컨테이너(tomcat), 쓰레드풀, DB(Hikari)기 종료되는 것을 볼 수 있다.

2024:09:01 00:00:38.729 INFO  [SpringContextShutdownHook] o.s.m.s.b.SimpleBrokerMessageHandler : Stopping...2024:09:01 00:00:38.729 INFO  [SpringContextShutdownHook] o.s.m.s.b.SimpleBrokerMessageHandler : BrokerAvailabilityEvent[available=false, SimpleBrokerMessageHandler [DefaultSubscriptionRegistry[cache[0 destination(s)], registry[0 sessions]]]]
2024:09:01 00:00:38.729 INFO  [SpringContextShutdownHook] o.s.m.s.b.SimpleBrokerMessageHandler : Stopped.
2024:09:01 00:00:38.730 INFO  [SpringContextShutdownHook] o.s.b.w.e.tomcat.GracefulShutdown : Commencing graceful shutdown. Waiting for active requests to complete2024:09:01 00:00:42.743 INFO  [tomcat-shutdown] o.s.b.w.e.tomcat.GracefulShutdown : Graceful shutdown complete
2024:09:01 00:00:42.778 INFO  [SpringContextShutdownHook] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.2024:09:01 00:00:43.677 INFO  [SpringContextShutdownHook] o.s.a.r.l.SimpleMessageListenerContainer : Successfully waited for workers to finish.
2024:09:01 00:00:43.688 INFO  [SpringContextShutdownHook] o.s.s.c.ThreadPoolTaskExecutor : Shutting down ExecutorService 'brokerChannelExecutor'2024:09:01 00:00:43.689 INFO  [SpringContextShutdownHook] o.s.s.c.ThreadPoolTaskScheduler : Shutting down ExecutorService 'messageBrokerTaskScheduler'
2024:09:01 00:00:43.690 INFO  [SpringContextShutdownHook] o.s.s.c.ThreadPoolTaskExecutor : Shutting down ExecutorService 'watcherExecutor'
2024:09:01 00:00:43.690 INFO  [SpringContextShutdownHook] o.s.s.c.ThreadPoolTaskExecutor : Shutting down ExecutorService 'beelineExecutor'
2024:09:01 00:00:43.730 INFO  [SpringContextShutdownHook] o.s.s.c.ThreadPoolTaskExecutor : Shutting down ExecutorService 'clientOutboundChannelExecutor'
2024:09:01 00:00:43.730 INFO  [SpringContextShutdownHook] o.s.s.c.ThreadPoolTaskExecutor : Shutting down ExecutorService 'clientInboundChannelExecutor'
2024:09:01 00:00:43.765 INFO  [SpringContextShutdownHook] o.s.o.j.LocalContainerEntityManagerFactoryBean : Closing JPA EntityManagerFactory for persistence unit 'default'
2024:09:01 00:00:43.776 INFO  [SpringContextShutdownHook] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Shutdown initiated...2024:09:01 00:00:43.828 INFO  [SpringContextShutdownHook] com.zaxxer.hikari.HikariDataSource : HikariPool-1 - Shutdown completed.

스프링 래빗엠큐의 종료

래빗엠큐의 종료가 시작되면,

메세지는 더 이상 수신하지 않고(SimpleBrokerMessageHandler 에 의해)
컨슈머가 이미 수신한 메세지 처리를 완료할 수 있도로 기다려준다. 기다리는 최대 시간이 Shutdown Timeout 이다.

# 코드 참고 : https://github.com/spring-projects/spring-amqp/blob/1f234cd0ff3a0166c8c69e38cc3f7b3167a90307/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/SimpleMessageListenerContainer.java#L688 

// 생략..
Runnable awaitShutdown = () -> {
  logger.info("Waiting for workers to finish.");
  try {
    boolean finished = this.cancellationLock.await(getShutdownTimeout(), TimeUnit.MILLISECONDS);    if (finished) {
      logger.info("Successfully waited for workers to finish.");
    }
    else {
      logger.info("Workers not finished.");
      if (isForceCloseChannel() || this.stopNow.get()) {
        canceledConsumers.forEach(consumer -> {
          if (logger.isWarnEnabled()) {
            logger.warn("Closing channel for unresponsive consumer: " + consumer);
          }
          consumer.stop();
        });
      }
    }
  }
  catch (InterruptedException e) {
    Thread.currentThread().interrupt();
    logger.warn("Interrupted waiting for workers.  Continuing with shutdown.");
  }
// 생략..

Default Shutdown Timeout 은 5초이다. 위 코드에서 getShutdownTimeout()으로 획득되는 값이다.

# 참고 : https://docs.spring.io/spring-amqp/reference/amqp/containerAttributes.html#shutdownTimeout 

When a container shuts down (for example, if its enclosing ApplicationContext is closed), 
it waits for in-flight messages to be processed up to this limit. 
Defaults to five seconds.

스프링 래빗엠큐 셧다운 타임아웃 시간 변경

5초 안에 대부분의 작업이 완료되지만, 그렇지 못한 특이 상황(비즈로직, 서버스펙, 외부연동 지연, 재시도 지연 등)도 있을 수 있으므로 시간을 변경하는 방법을 알아본다.

코틀린

@Bean
fun rabbitListenerContainerFactory(
    connectionFactory: ConnectionFactory,
    configurer: SimpleRabbitListenerContainerFactoryConfigurer
): SimpleRabbitListenerContainerFactory {
    val factory = SimpleRabbitListenerContainerFactory()
    configurer.configure(factory, connectionFactory)
    factory.setContainerCustomizer { c: SimpleMessageListenerContainer ->
        c.setShutdownTimeout(10000) # 10초로 변경    }
    return factory
}

자바

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(final ConnectionFactory connectionFactory,
     final SimpleRabbitListenerContainerFactoryConfigurer configurer) {
  SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
  configurer.configure(factory, connectionFactory);
  factory.setContainerCustomizer(c -> c.setShutdownTimeout(10000L)); # 10초로 변경  return factory;
}

기타

ChatGPT 4o가 아래와 같이 알려주는데 동작안한다. (24년 9월 1일 기준)

# 동작하지 않는 코드
## springVersion 2.3.12.RELEASE
## org.springframework.amqp:spring-rabbit:2.2.18.RELEASE
@Bean
fun container(connectionFactory: ConnectionFactory): SimpleMessageListenerContainer {
  val container = SimpleMessageListenerContainer(connectionFactory)
  container.setShutdownTimeout(30000) // 5초 동안 graceful shutdown 대기
  return container
}