RabbitMQ Broken pipe connection closed

메시지를 받아서 전달해 주는 오픈소스로 RabbitMQ를 많이 사용한다.

RabbitMQ 운용 중에 발생하는 상황에 대해서 정리해봅니다.

 

문제

Application에서 message publish할 때 아래와 같은 오류가 Application log에 남고 RabbitMQ와의 연결이 끊어집니다.

큐에 있던 consumer 들이 모두 제거되서 메시지를 처리하지 못하는 상황이 발생합니다.

Caused by: java.net.SocketException: Broken pipe   
at java.net.SocketOutputStream.socketWrite0(Native Method) ~[?:1.8.0_73]

 

RabbitMQ 로그에는 아래와 같은 로그가 남습니다.

{writer,send_failed,{error,timeout}}

 

원인

Spring Cloud Stream 오픈소스를 쓰면서 acknowledge mode를 none으로 변경했고 해당 설정은 RabbitMQ의 acknowledge mode가 auto인 것으로 설정이 됩니다.

RabbitMQ acknowledge mode auto 설정은 메시지에 대한 클라이언트 ack 없이 전송 후에 바로 큐에서 삭제합니다.

안정성보다 속도를 중시하는 설정이었고 클라이언트의 처리 속도와 관계없이 RabbitMQ에선 계속 메시지를 보냅니다.

초기 처리량에 대한 고민으로 적용했던 부분인데 그게 문제가 되었네요

Pruducer의 메시지 발행 속도가 Consumer의 처리 속도보다 월등히 높았고 결국 TCP 버퍼 용량이 차서 위와 같이 RabbitMQ 로그에 쓰기 실패 오류가 나고 timeout으로 인해 connection close가 발생합니다. 

 

해결

우선 spring cloud stream의 AcknowlegeMode 클래스를 보면 다음과 같습니다.

public enum AcknowledgeMode {

	/**
	 * No acks - {@code autoAck=true} in {@code Channel.basicConsume()}.
	 */
	NONE,

	/**
	 * Manual acks - user must ack/nack via a channel aware listener.
	 */
	MANUAL,

	/**
	 * Auto - the container will issue the ack/nack based on whether
	 * the listener returns normally, or throws an exception.
	 * <p><em>Do not confuse with RabbitMQ {@code autoAck} which is
	 * represented by {@link #NONE} here</em>.
	 */
	AUTO;

spring cloud stream 기본값은 Auto이지만 RabbitMQ의 manual과 같고 NONE으로 설정했을 때 autoAck 설정이 됩니다.

위 주석과 같이 조금 혼동되지만 꼭 기억하세요

Spring Cloud Stream NONE == RabbitMQ AUTO

우선 acknowledge mode를 spring cloud stream 기본값인 auto로 바꿔줬고 consumer 개수와 prefetch count를 적절히 조정했습니다.

 

Consumer와 Prefetch Count에 대해서

저희 환경에서 consumer 개수는 서버 대수와 서버 CPU 리소스(core 개수)를 고려해서 정했고 타 서비스와 연동하는 구간이 있어서 대기 시간을 고려하여 그보다 조금 더 많은 쓰레드 개수를 선택했습니다.

AMQP 0-9-1 는 아래와 같이 안 읽은 메시지 최대 개수를 제한할 수 있도록 basic.qos 메서드에 파라미터로 지정해서 사용합니다.

channel.basicQos(prefetchCount);

prefetch count는 spring cloud stream 2.0부터 기본 설정이 250개로 변경되었고 1로 지정하면 확실한 순서보장을 할 수 있습니다.

prefetch count를 10으로 하고 서버 10대에 consumer를 4개로 지정하면

RabbitMQ Management에 보이는 Unacked 개수는 최대 400까지가 나옵니다.

그 이상은 RabbitMQ에서 이전 메시지에 대한 ack를 받지 않는 한 전달을 하지 않습니다.

이렇게 되면 메모리가 부족하거나 영속적 메시지인 경우 디스크에 메시지를 바로 쓰면서

디스크 공간(disk space)가 부족한 상황이 발생합니다.

이 부분을 주의해서 메시지 Ready 건 수와 메모리 사용량 등 RabbitMQ 리소스를 모니터링 하셔야 됩니다.

 

설정에 대한 테스트를 해보겠습니다.

아래는 autoAck로 설정한 화면입니다. 계속 메시지를 발행했고 Consumer는 임의로 sleep을 줘서 지연 시켰습니다.

autoAck

위 지표 중에 Ready에 있는 건 RabbitMQ가 가지고 있는 메시지 개수이고 Unacked는 메시지를 전달했지만 ack를 받지 못한 메시지 개수입니다.

Consumer utilisation은 큐가 consumer에게 즉시 메시지를 전달할 수 있는 시간 비율인데

이것이 100%보다 적으면 consumer를 늘리거나 메시지를 consumer가 처리 하는 중 지연이 있는 부분을 제거해서 더 빠르게 하거나

더 높은 prefetch count를 설정합니다.

Consumer utilisation

위 처럼 해당 [?] 를 클릭하면 자세한 정보를 볼 수 있으니 RabbitMQ Management를 잘 활용하시는 것도 좋습니다.

메시지 [?] 설명대로 영속적인(persistent) 메시지는 디스크와 메모리에 있을 수 있고 일시적인(transient) 메시지도 메모리가 부족하면 page out 되서 디스크에 있을 수 있습니다.

아래는 Consumer 상태입니다. Ack required가 false여서 O 표시로 되어 있고 autoAck인 경우 prefetch count 설정이 무시됩니다.

autoAck

 

아래는 manualAck로 설정된 상태입니다.

1대의 애플리케이션에서 1개의 consumer, prefetch count 50으로 설정했고 최대 50개의 unacked 메시지를 RabbitMQ에서 클라이언트로 전달합니다.

manual ack

아래와 같이 manual ack로 설정되었고 prefetch count 50이 정상적으로 보입니다.

manualAck

 

마무리

spring cloud stream의 ack mode를 none으로 바꾸는 설정은 좀 더 신중해야 합니다.

consumer 처리 속도를 보장하고 메시지 유실에 대한 염려가 없을 때 성능에 이점이 있으면 그때 설정하는 것이 맞습니다.

consumer 개수 설정과 prefetch count 조정을 통해 더 나은 애플리케이션을 만드세요~!!

 

참고

help.mulesoft.com/s/article/RabbitMQ-resets-the-connection-and-messages-are-resent-to-AMQP-inbound-endpoint

RabbitMQ resets the connection and messages are resent to AMQP inbound endpoint

www.cloudamqp.com/blog/part4-rabbitmq-13-common-errors.html

13 Common RabbitMQ Mistakes and How to Avoid Them - CloudAMQP

www.cloudamqp.com/blog/part1-rabbitmq-best-practice.html

Part 1: RabbitMQ Best Practices - CloudAMQP