SpringBoot Amqp RabbitAdmin Sınıfı

declareBinding metodu

Örnek

Şöyle yaparız

@Autowired
private RabbitAdmin rabbitAdmin;

public void addNewQueue(String queueName, String exchangeName, String routingKey) {
  Queue queue = new Queue(queueName, true, false, false);
  Binding binding = new Binding(
    queueName,
    Binding.DestinationType.QUEUE,
    exchangeName,
    routingKey,
    null
   );
  rabbitAdmin.declareQueue(queue);
  rabbitAdmin.declareBinding(binding);
}

deleteQueue metodu

Örnek

Şöyle yaparız

String queueName = ...
rabbitAdmin.deleteQueue(queueName);

SpringBoot Amqp RabbitMQ Kullanımı - DirectExchange

Giriş

Direct Exchange'e bir kuyruk ya da birden fazla kuyruk takılabilir. 

Mesaj gönderirken kuyruk ismi routing key olarak belirtilir.

Örnek - DirectExchange  ve Tek Kuyruk

Şöyle yaparız. Burada iki tane farklı DirectExchange yaratılıyor.

@Configuration
public class RabbitConfigure {

  // build queue
  @Bean
  public Queue helloQuery() {
    return new Queue("hello");// Pass ordinary string message
  }

  @Bean
  public Queue userQuery() {
    return new Queue("user");// Pass java object message
  }
}

Mesaj göndermek için şöyle yaparız. convertAndSend() metoduna exchange ismi veriliyor.

@Component
public class HelloSender {

  @Autowired
  public RabbitTemplate rabbitTemplate;

  //sent to hello queue
  public void sendHello() {
    String message = "...";
    rabbitTemplate.convertAndSend("hello", message);
  }

  //Send to user queue
  public void sendUser() {
    User user = new User();
    ...
    rabbitTemplate.convertAndSend("user", user);
  }
}

Örnek - DirectExchange  ve Tek Kuyruk

Küçük gemilerin ana gemiye mesaj göndermesi durumunda kullanılabilir. Küçük gemi için elimizde şöyle bir kod olsun

## application.yml for the ship's application
## have to change property values for each ship
ship:
  name: rocinante
  update-freq: 1000

broker:
  exchange:
    direct:
      ship:
        name: rocinante-direct-exchange
        routing-key: __rocinante 
      station:
        name: tyco-direct-exchange 
        routing-key: __scheduled-update
    fanout:
      name: tyco-fanout-exchange
    queue:
      name: rocinante

Küçük gemi kodu şöyledir

@Component
@EnableScheduling
public class UpdateScheduler {

  @Value("${ship.name}")
  private String shipName;
  @Value("${broker.exchange.direct.station.name}")
  private String directExchange; //tyco-direct-exchange

  @Value("${broker.exchange.direct.station.routing-key}")
  private String directExchangeRoutingKey; //__scheduled-update

  private Long shipUpdateFrequency;

  @Value("${ship.update-freq}")
  private void setShipUpdateFrequency(String frequency) {
    this.shipUpdateFrequency = Long.parseLong(frequency);
  }

  @Autowired
  private final RabbitTemplate rabbitTemplate;


  @SneakyThrows
  @Scheduled(fixedDelay = 1)
  public void sendUpdates() {
    String updateMessage = shipName + ": Update at " + new Date() + " " +
      ParameterFactory.getParameter();
    rabbitTemplate.convertAndSend(directExchange, directExchangeRoutingKey, updateMessage);
    Thread.sleep(shipUpdateFrequency);
  }
} 

Ana gemi yaml dosyası şöyledir

## application.yml for the station's application
station:
  name: Tyco

broker:
  exchange:
    direct:
      name: tyco-direct-exchange
      routing-key: __scheduled-update
      queue:
        auto-queue: auto-queue
    fanout:
      name: tyco-fanout-exchange

Ana gemi broker configuration bean'leri şöyledir

@Configuration
public class BrokerConfiguration {
  static String directExchangeQueue;
  static String directExchange;
  static String directRoutingKey;
  @Value("${broker.exchange.direct.routing-key}")
  private void setDirectRoutingKey(String routingKey) {
    BrokerConfiguration.directRoutingKey = routingKey;
  }
  @Value("${broker.exchange.direct.name}")
  private void setDirectExchange(String exchangeName) {
    BrokerConfiguration.directExchange = exchangeName;
  }

  @Value("${broker.exchange.direct.queue.auto-queue}")
  private void setQueueName(String queueName) {
    BrokerConfiguration.directExchangeQueue = queueName;
  }
  @Bean
  DirectExchange directExchange() {
    return new DirectExchange(BrokerConfiguration.directExchange);
  }

  @Bean
  Queue directExchangeQueue() {
    return new Queue(BrokerConfiguration.directExchangeQueue);
  }

  @Bean
  Binding updateQueueBinding(Queue directExchangeQueue, DirectExchange directExchange) {
    return BindingBuilder
      .bind(directExchangeQueue)
      .to(directExchange)
      .with(BrokerConfiguration.directRoutingKey);
  }
}

Mesajları dinleyen kod şöyledir

// message listener configuration
@Configuration
public class MessageListenerConfiguration {
  @Autowired
  private final BrokerConfiguration brokerConfiguration;

  @Bean
  MessageListenerAdapter listenerAdapter(MessageHandler messageHandler) {
    return new MessageListenerAdapter(messageHandler, "receiveMessage");
  }

  @Bean
  SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
                                          MessageListenerAdapter listenerAdapter) {
    SimpleMessageListenerContainer container = new  SimpleMessageListenerContainer();
    container.setConnectionFactory(connectionFactory);
    container.setQueueNames(brokerConfiguration.directExchangeQueue);
    container.setMessageListener(listenerAdapter);
    return container;
  }
}

@Component
public class MessageHandler {
  // Callback method to handle the recived messages
  public void receiveMessage(String message) {
    System.out.println("> " + message);
  }
}

Örnek - DirectExchange  ve İki Kuyruk

Şöyle yaparız. Burada yine tek bir DirectExchange var ancak bu sefer iki tane queue bağlanıyor.

@Configuration
public class RabbitMqConfiguration {
  @Value("${myapp.rabbitmq.exchange-name}")
  private String exchangeName;
  @Bean
  Queue someQueue() {
    return new Queue("someQueue", false);
  }
  @Bean
  Queue anotherQueue() {
    return new Queue("anotherQueue", false);
  }
  @Bean
  public DirectExchange directExchange() {
    return new DirectExchange(exchangeName);
  }
  @Bean
  Binding someBinding(final Queue someQueue, DirectExchange exchange) {
    return BindingBuilder.bind(someQueue).to(exchange).with("some-router");
  }
  @Bean
  Binding anotherBinding(final Queue anotherQueue, DirectExchange exchange) {
    return BindingBuilder.bind(anotherQueue).to(exchange).with("another-router");
  }
}

Routing key değerine göre istenilen kuyruğa göndermek için şöyle yaparız

@Service
public class SomeJobProducer {

  private final RabbitTemplate rabbitTemplate;

  @Value("${myapp.rabbitmq.exchange-name}")
  private String exchangeName;

  public void sendToQueue(String workId) {
    rabbitTemplate.convertAndSend(exchangeName, "some-router", workId);
  }
}

@Service
public class AnotherJobProducer {

  private final RabbitTemplate rabbitTemplate;

  @Value("${myapp.rabbitmq.exchange-name}")
  private String exchangeName;

  public void sendToQueue(String workId) {
    rabbitTemplate.convertAndSend(exchangeName, "another-router", workId);
  }
}

Listener'ların doğru kuyruğu dinlemesi gerekir. Şöyle yaparız

@Service
public class SomeQueueListener {

  @RabbitListener(queues = "someQueue")
  public void handleSomeJob(String workId) {
    System.out.printf("Working on SomeQueue... %s%n", workId);
  }
}
@Service
public class AnotherQueueListener {

  @RabbitListener(queues = "anotherQueue")
  public void handleAnotherJob(String workId) {
    System.out.printf("Working on anotherQueue... %s%n", workId);
  }
}

SpringBoot Amqp DirectRabbitListenerContainerFactory Sınıfı

setConnectionFactory metodu

Şöyle yaparız

@Bean
public CachingConnectionFactory cachingConnectionFactory(){
  CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
  cachingConnectionFactory.setAddresses("10.10.121.199:35682");
  cachingConnectionFactory.setUsername("guest");
  cachingConnectionFactory.setPassword("guest");
  return cachingConnectionFactory;
}

@Bean
public DirectRabbitListenerContainerFactory directFactory(ConnectionFactory cf) {
  DirectRabbitListenerContainerFactory factory = new DirectRabbitListenerContainerFactory();
  factory.setConnectionFactory(cf);
  return factory;
}

SpringBoot Amqp RabbitMQ Kullanımı

Giriş

RabbitMQ kullanırken hem Producer hem de Consumer tarafından durable olmayan kuyruk'ları yaratan kodlar bulunur. Bunun sebebi önce Consumer çalışırsa, kuyruk yok hatası almamaktır. Eğer kuyruk zaten mevcutsa, Consumer tarafından tekrar yaratılmaya çalışmasının bir zararı/etkisi yok. Sadece boş işlem gibi düşünülebilir.

Maven

Şu satırı dahil ederiz. Eğer reactive çalışmak istersek "reactor-rabbitmq" projesi de var.

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

Önemli Sınıflar

ConnectionFactory 

RabbitAdmin 

Jackson2JsonMessageConverter

RabbitListenerConfigurer 

Queue 

Binding 

Producer İçin Önemli Sınıflar

Jackson2JsonMessageConverter

RabbitTemplate 

Consumer İçin Önemli Sınıflar

MappingJackson2MessageConverter

RabbitListenerEndpointRegistry 

DefaultMessageHandlerMethodFactory 

Ayarlar

Bağlantı ayarları için SpringBoot spring.rabbitmq Ayarları yazısına bakınız

Mesaj Göndermek

RabbitTemplate sınıfının convertAndSend() metodu kullanılır. Kullanılacak serialization yöntemini RabbitTemplate.setMessageConverter() çağrısı ile belirlemek gerekir. Eğer bir MessageConverter atanmadıysa, RabbitTemplate Java Binary Serialization yöntemini kullanır.

Mesaj Almak

@RabbitListener Anotasyonu ile mesajı alacağımız kuyruk belirtilir.

AmqpAdmin Arayüzü

Bu arayüzün tam yolu org.springframework.amqp.core.AmqpAdmin şeklinde. Kuyruk yaratmak için kullanılan arayüz. Açıklaması şöyle

The AMQP specification describes how the protocol can be used to configure queues, exchanges, and bindings on the broker. These operations (which are portable from the 0.8 specification and higher) are present in the AmqpAdmin interface in the org.springframework.amqp.core package. The RabbitMQ implementation of that class is RabbitAdmin located in the org.springframework.amqp.rabbit.core package.

Kuyruk - Queue

org.springframework.amqp.core.Queue sınıfından bir kuyruk yaratılmalıdır. Queue sınıfı direkt yaratılabileceği gibi, QueueBuilder sınıfı da kullanılabilir.

DirectExchange

DirectExchange yazısına taşıdım

Örnek - FanoutExchange

FanoutExchange yazısına taşıdım

Örnek - TopicExchange

Şöyle yaparız

@Configuration
public class TopicConfigure {

  @Bean
  public Queue topicMessageA() {
    return new Queue("topic.messageA");
  }

  @Bean
  public Queue topicMessageB() {
    return new Queue("topic.messageB");
  }

  @Bean
  public TopicExchange exchangeTopic() {
    return new TopicExchange("topicExchange");
  }

  //The matching rule set here is topic.messageA
  @Bean
  public Binding bindingExchangeMessageA(TopicExchange exchange) {
    return BindingBuilder.bind(topicMessageA()).to(exchange).with("topic.messageA");
  }
  //The matching rule here is topic.[any word]
  @Bean
  public Binding bindingExchangeMessageB(TopicExchange exchange) {
    return BindingBuilder.bind(topicMessageB()).to(exchange).with("topic.#");
  }
}   

Mesaj gönder göndermek için şöyle yaparız

@Component
public class TopicSender {

  @Autowired
  public RabbitTemplate rabbitTemplate;

  public void sendTopicA() {
    String message = "...";
    rabbitTemplate.convertAndSend("topicExchange", "topic.messageA", message);
  }

  public void sendTopicB() {
    String message = "...";
    rabbitTemplate.convertAndSend("topicExchange", "topic.messageB", message);
  }
}      

Örnek - TopicExchange

Eğer istenirse farklı bir exchange de kullanılabilir. Şöyle yaparız

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;

static final String topicExchangeName = "spring-boot-exchange";

static final String queueName = "spring-boot";

@Bean
Queue queue() {
  return new Queue(queueName, false);
}

@Bean
TopicExchange exchange() {
  return new TopicExchange(topicExchangeName);
}

@Bean
Binding binding(Queue queue, TopicExchange exchange) {
  return BindingBuilder.bind(queue).to(exchange).with("foo.bar.#");
}

Örnek - headersExchange

Şöyle yaparız

@Configuration
public class HeaderConfigure {

  @Bean
  public Queue headerQueue() {
    return new Queue("header");
  }

  @Bean
  public HeadersExchange headersExchange() {
    return new HeadersExchange("headersExchange");
  }

  // Here is bound to the key value pair key = color value = red
  // Of course you can bind multiple sets of key-value pairs
  @Bean
  HeadersExchangeMapBindingCreator headersBinding(HeadersExchange headersExchange) {
    Map<String, Object> map = new HashMap<>();
    map.put("color", "red");
    return BindingBuilder.bind(headerQueue()).to(headersExchange).whereAll(map);
    //return BindingBuilder.bind(headerQueue()).to(headersExchange).where("color")
    //  .matches("red");
  }
} 

Şöyle yaparız

@Component
public class HeadersSender {

  @Autowired
  private RabbitTemplate rabbitTemplate;

  // Set the key value pair of the message header through MessageProperties
  // After conversion through SimpleMessageConverter ()
  // If you pass the java object, you need to use Jackson2JsonMessageConverter ()
  public void sendHeaders() {
    String str = "...";
    MessageProperties messageProperties = new MessageProperties();
    messageProperties.setHeader("color", "red");
    MessageConverter messageConverter = new SimpleMessageConverter();
    Message message = messageConverter.toMessage(str, messageProperties);
    rabbitTemplate.convertAndSend("headersExchange", "", message);
  }

  public void sendWrongHeaders() {
    String str = "...";
    MessageProperties messageProperties = new MessageProperties();
    messageProperties.setHeader("color", "gold");
    MessageConverter messageConverter = new SimpleMessageConverter();
    Message message = messageConverter.toMessage(str, messageProperties);
    rabbitTemplate.convertAndSend("headersExchange", "", message);
  }
}    

Eğer Kuyruk Broker'da Zaten Varsa

Kuyruğu tekrar yaratmaya çalışmanın bir etkisi yok. Açıklaması şöyle

So, you should declare Queue, Exchange and Binding beans in your application context and AmqpAdmin will take care about their definition on the target Broker.

There must be a note that according AMQP protocol, if entity already exists on the Broker, the declaration is just silent and idempotent.

So, in your case you don't need to worry about queues existence and just provide their declarations as beans in the application context.

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;
  }
  ...
}