《RabbitMQ》什麼是死信隊列

一 什麼是死信隊列

當一條消息在隊列中出現以下三種情況的時候,該消息就會變成一條死信。

  • 消息被拒絕(basic.reject / basic.nack),並且requeue = false
  • 消息TTL過期
  • 隊列達到最大長度

當消息在一個隊列中變成一個死信之後,如果配置了死信隊列,它將被重新publish到死信交換機,死信交換機將死信投遞到一個隊列上,這個隊列就是死信隊列。

二 實現死信隊列

2.1 原理圖

死信交換機.png

2.2 創建消費者

創建一個消費者,綁定消費隊列及死信交換機,交換機默認為direct模型,死信交換機也是,arguments綁定死信交換機和key。(註解支持的具體參數文末會附上)

public class DirectConsumer {
    @RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = "javatrip",arguments = 
                    {@Argument(name="x-dead-letter-exchange",value = "deadExchange"),
                     @Argument(name="x-dead-letter-routing-key",value = "deadKey")
                    }),
                    exchange = @Exchange(value="javatripDirect"),
                    key = {"info","error","warning"}
            )
    })
public void receive1(String message, @Headers Map<String,Object> headers, Channel channel)throws Exception{
    System.out.println("消費者1"+message);
}

2.3 創建生產者

public void publishMessage(String message){

    rabbitTemplate.setMandatory(true);
    rabbitTemplate.convertAndSend("javatripDirect","info",message);
}

三 造成死信的三種情況

3.1 拒絕消息,並且禁止重新入隊

  1. 設置yml為手動簽收模式
spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual
  1. 設置拒絕消息並禁止重新入隊
Long deliverTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
channel.basicNack(deliverTag,false,false);
  1. 綁定死信隊列
@RabbitListener(bindings = {
    @QueueBinding(
        value = @Queue(value = "javatripDead"),
        exchange = @Exchange(value = "deadExchange"),
        key = "deadKey"
    )
})
public void receive2(String message){
    System.out.println("我是一條死信:"+message);
}

3.2 消息TTL過期

綁定業務隊列的時候,增加消息的過期時長,當消息過期後,消息將被轉發到死信隊列中。

@RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = "javatrip",arguments =
                    {@Argument(name="x-dead-letter-exchange",value = "deadExchange"),
                     @Argument(name="x-dead-letter-routing-key",value = "deadKey"),
                     @Argument(name = "x-message-ttl",value = "3000")
                    }),
                    exchange = @Exchange(value="javatripDirect"),
                    key = {"info","error","warning"}
            )
    })
public void receive1(String message, @Headers Map<String,Object> headers, Channel channel)throws Exception{
    System.out.println("消費者1"+message);
}

3.3 隊列達到最大長度

設置消息隊列長度,當隊列中的消息達到最大長度後,繼續發送消息,消息將被轉發到死信隊列中。

@RabbitListener(bindings = {
            @QueueBinding(value = @Queue(value = "javatrip",arguments =
                    {@Argument(name="x-dead-letter-exchange",value = "deadExchange"),
                     @Argument(name="x-dead-letter-routing-key",value = "deadKey"),
                     @Argument(name = "x-max-length",value = "3")
                    }),
                    exchange = @Exchange(value="javatripDirect"),
                    key = {"info","error","warning"}
            )
    })
public void receive1(String message, @Headers Map<String,Object> headers, Channel channel)throws Exception{
    System.out.println("消費者1"+message);
}

四 Spring Boot整合RabbitMQ用到的幾個註解

  1. @QueueBinding作用就是將隊列和交換機進行綁定,主要有以下三個參數:
@Target({})
@Retention(RetentionPolicy.RUNTIME)
public @interface QueueBinding {

	/**
	 * @return the queue.
	 */
	Queue value();

	/**
	 * @return the exchange.
	 */
	Exchange exchange();

	/**
	 * @return the routing key or pattern for the binding.
	 * Multiple elements will result in multiple bindings.
	 */
	String[] key() default {};
}
  1. @Queue是聲明隊列及隊列的一些屬性,主要參數如下:
@Target({})
@Retention(RetentionPolicy.RUNTIME)
public @interface Queue {

	/**
	 * @return the queue name or "" for a generated queue name (default).
	 */
	@AliasFor("name")
	String value() default "";

	/**
	 * @return the queue name or "" for a generated queue name (default).
	 * @since 2.0
	 */
	@AliasFor("value")
	String name() default "";

	/**
	 * 是否持久化
	 */
	String durable() default "";

	/**
	 * 是否獨享、排外的.
	 */
	String exclusive() default "";

	/**
	 * 是否自動刪除;
	 */
	String autoDelete() default "";

	/**
	 * 隊列的其他屬性參數
	 * (1)x-message-ttl:消息的過期時間,單位:毫秒;
         *(2)x-expires:隊列過期時間,隊列在多長時間未被訪問將被刪除,單位:毫秒;
         *(3)x-max-length:隊列最大長度,超過該最大值,則將從隊列頭部開始刪除消息;
         *(4)x-max-length-bytes:隊列消息內容佔用最大空間,受限於內存大小,超過該閾值則從隊列頭部開始刪除消息;
         *(5)x-overflow:設置隊列溢出行為。這決定了當達到隊列的最大長度時消息會發生什麼。有效值是drop-head、reject-publish或reject-publish-dlx。仲裁隊列類型僅支持drop-head;
         *(6)x-dead-letter-exchange:死信交換器名稱,過期或被刪除(因隊列長度超長或因空間超出閾值)的消息可指定發送到該交換器中;
         *(7)x-dead-letter-routing-key:死信消息路由鍵,在消息發送到死信交換器時會使用該路由鍵,如果不設置,則使用消息的原來的路由鍵值
         *(8)x-single-active-consumer:表示隊列是否是單一活動消費者,true時,註冊的消費組內只有一個消費者消費消息,其他被忽略,false時消息循環分發給所有消費者(默認false)
         *(9)x-max-priority:隊列要支持的最大優先級數;如果未設置,隊列將不支持消息優先級;
         *(10)x-queue-mode(Lazy mode):將隊列設置為延遲模式,在磁盤上保留儘可能多的消息,以減少RAM的使用;如果未設置,隊列將保留內存緩存以儘可能快地傳遞消息;
         *(11)x-queue-master-locator:在集群模式下設置鏡像隊列的主節點信息。
	 */
	Argument[] arguments() default {};
}
  1. @Exchange是聲明交換及交換機的一些屬性,
@Target({})
@Retention(RetentionPolicy.RUNTIME)
public @interface Exchange {

	String TRUE = "true";

	String FALSE = "false";

	/**
	 * @return the exchange name.
	 */
	@AliasFor("name")
	String value() default "";

	/**
	 * @return the exchange name.
	 * @since 2.0
	 */
	@AliasFor("value")
	String name() default "";

	/**
	 * 交換機類型,默認DIRECT
	 */
	String type() default ExchangeTypes.DIRECT;

	/**
	 * 是否持久化
	 */
	String durable() default TRUE;

	/**
	 * 是否自動刪除
	 */
	String autoDelete() default FALSE;

	/**
	 * @return the arguments to apply when declaring this exchange.
	 * @since 1.6
	 */
	Argument[] arguments() default {};
}