왜 spring-amqp 소비자 성능이 매우 느 립니까?

Question

저는 프로듀서와 소비자를 동시에 시작했습니다. 6 시간 후 생산자는 약 6 개의 crores 메시지를 대기열로 보내고 6 시간 후에 생산자가 멈추었지만 소비자는 18 시간 후에도 4 개의 crores 메시지가 대기열에 계속 남아 있습니다. 소비자 성능이 매우 느린 이유를 알려주시겠습니까?

미리 감사드립니다.

@Bean 
    public SimpleMessageListenerContainer listenerContainer() { 
     SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); 
     container.setConnectionFactory(connectionFactory()); 
     container.setQueueNames(this.queueName); 
     container.setMessageListener(new MessageListenerAdapter(new TestMessageHandler(), new JsonMessageConverter())); 
     return container; 
    } 
@Bean 
    public ConnectionFactory connectionFactory() { 
     CachingConnectionFactory connectionFactory = new CachingConnectionFactory(
       "localhost"); 
     connectionFactory.setUsername("guest"); 
     connectionFactory.setPassword("guest"); 
     return connectionFactory; 
    } 

    @Bean 
    public RabbitTemplate rabbitTemplate() { 
     RabbitTemplate template = new RabbitTemplate(connectionFactory()); 
     template.setMessageConverter(new JsonMessageConverter()); 
     template.setRoutingKey(this.queueName); 
     template.setQueue(this.queueName); 
     return template; 
    } 

    public class TestMessageHandler { 
      // receive messages 
     public void handleMessage(MessageBeanTest msgBean) { 
        // Storing bean data into CSV file 
      } 
    } 

Answer 1

WikiPedia에 따르면, crore == 10,000,000이므로 6 천만을 의미합니다.

컨테이너는 청취자만큼 빠른 메시지 처리 만 할 수 있습니다. 각 메시지로 수행중인 작업을 분석해야합니다.

최적의 성능을 얻으려면 컨테이너 동시성 설정 (concurrentConsumers), 프리 페치 (prefetch) 등을 시험해야하지만 대부분의 처리 시간을 필요로하는 청취자가됩니다. 컨테이너는 매우 쓰레기 오버 헤드가 있습니다. 리스너가 제대로 구성되지 않은 경우 병행 성을 높이면 도움이되지 않습니다.

트랜잭션을 사용하는 경우 소비가 크게 줄어 듭니다.

메시지와 아무런 관련이없는 리스너를 사용해보십시오.

마지막으로, 이와 같은 질문을 할 때 항상 구성을 표시해야합니다.

Answer 2

Gary의 제안에 따라 다음과 같이 설정할 수 있습니다. 체크 아웃 @ rabbitListener

@Bean 
public SimpleRabbitListenerContainerFactory listenerContainer( { 
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); 
    factory.setConnectionFactory(baseConfig.connectionFactory()); 
    factory.setConcurrentConsumers(7); // choose a value 
    factory.setPrefetchCount(1); // how many messages per consumer at a time 
    factory.setMaxConcurrentConsumers(10); // choose a value 
    factory.setDefaultRequeueRejected(false); // if you want to deadletter 
    return factory; 
}