SpringBoot Amqp SimpleRabbitListenerContainerFactory Sınıfı

Giriş

Normalde bu sınıfı yaratmamıza gerek yok. Spring bizim için varsayılan davranışlara sahip bir nesne yaratıyor. Bu davranışları değiştirmek istersek bu sınıfı elle yaratmak gerekir.

Örnek

Bu sınıfı yaratırız

@Configuration
static class RabbitConfiguration {

  @Bean
  public SimpleRabbitListenerContainerFactory myFactory(
    SimpleRabbitListenerContainerFactoryConfigurer configurer) {
    
    SimpleRabbitListenerContainerFactory factory =
      new SimpleRabbitListenerContainerFactory();
    configurer.configure(factory, connectionFactory);
    factory.setMessageConverter(myMessageConverter());
    return factory;
  }
}

Daha sonra sınıfı @RabbitListener ile kullanırız. Şöyle yaparız

@Component
public class MyBean {

  @RabbitListener(queues = "someQueue", containerFactory="myFactory")
  public void processMessage(String content) {
    // ...
  }
}

DirectMessageListenerContainer İle Farkı Nedir?

Açıklaması şöyle. Sürüm 2.0'dan önce RabbitMQ Client Thread concurrent çalışmadığı için ara bir kuyruk kullanmak zorunda kalmışlar.

The SMLC has a dedicated thread for each consumer (concurrency) which polls an internal queue. When a new message arrives for a consumer on the client thread, it is put in the internal queue and the consumer thread picks it up and invokes the listener. This was required with early versions of the client to provide multi-threading. With the newer client that is not a problem so we can invoke the listener directly (hence the name).

Ancak sürüm 2.0'dan sonra DirectMessageListenerContainer kullanılabilir. Açıklaması şöyle

Version 2.0 introduced the DirectMessageListenerContainer (DMLC). Previously, only the SimpleMessageListenerContainer (SMLC) was available.

setConsumerTagStrategy metodu

Şöyle yaparız.

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
 SimpleRabbitListenerContainerFactoryConfigurer configurer,
 ConnectionFactory connectionFactory) {

  SimpleRabbitListenerContainerFactory f = new SimpleRabbitListenerContainerFactory();
  configurer.configure(f, connectionFactory);
  
  f.setConsumerTagStrategy(q -> "myConsumerFor." + q);
  return f;
}

setDefaultRequeueRejected metodu

Açıklaması şöyle

If retries are not enabled and the listener throws an exception, by default the delivery will be retried indefinitely. You can modify this behavior in two ways; set the defaultRequeueRejected property to false and zero re-deliveries will be attempted; or, throw an AmqpRejectAndDontRequeueException to signal the message should be rejected. This is the mechanism used when retries are enabled and the maximum delivery attempts are reached.

setErrorHandler metodu

Açıklaması şöyle. Bazı exception'lar fatal (ölümcül) ve tekrar kuyruğa koymaya gerek yok. Örneğin MessageConversionException ölümcül bir hata. Bu mesajın işlenmesi mümkün değil.

SimpleRabbitListenerContainerFactory by default uses ConditionalRejectingErrorHandler. 

Simply put, ConditionalRejectingErrorHandler decides whether to reject a specific message or not. When the message that caused an exception is rejected, it won’t be requeued.

Şöyle yaparız

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
 ConnectionFactory connectionFactory,
 SimpleRabbitListenerContainerFactoryConfigurer configurer) {

  SimpleRabbitListenerContainerFactory f = new SimpleRabbitListenerContainerFactory();
  configurer.configure(f, connectionFactory);
	  
 f.setErrorHandler(errorHandler());
 return f;
}
 
@Bean
public ErrorHandler errorHandler() {
  return new ConditionalRejectingErrorHandler(customExceptionStrategy());
}
 
@Bean
FatalExceptionStrategy customExceptionStrategy() {
  return new CustomFatalExceptionStrategy();
}

setMaxConcurrentConsumers metodu

Şöyle yaparız

@Configuration
@EnableRabbit
public class MessagingConfig {
  private static final int MAX_CONSUMERS = 2;
  private final ConnectionFactory connectionFactory;
  
  @Bean
  public AmqpAdmin amqpAdmin() {
    return new RabbitAdmin(connectionFactory);
  }
  
  @Bean(name = "invoiceProcessingRabbitListenerContainerFactory")
  SimpleRabbitListenerContainerFactory invoiceProcessingRabbitListenerContainerFactory() {
    SimpleRabbitListenerContainerFactory f = new SimpleRabbitListenerContainerFactory();
    f.setConnectionFactory(connectionFactory);
    f.setMaxConcurrentConsumers(MAX_CONSUMERS);
    return factory;
  }
  ...
}