RabbitMQ整合Spring Booot【死信隊列】

Config:

import java.util.HashMap;
import java.util.Map;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

//Fanout 類型 發佈訂閱模式
@Component
public class FanoutConfig {

    /**
     * 定義死信隊列相關信息
     */
    public final static String deadQueueName = "dead_queue";
    public final static String deadRoutingKey = "dead_routing_key";
    public final static String deadExchangeName = "dead_exchange";
    /**
     * 死信隊列 交換機標識符
     */
    public static final String DEAD_LETTER_QUEUE_KEY = "x-dead-letter-exchange";
    /**
     * 死信隊列交換機綁定鍵標識符
     */
    public static final String DEAD_LETTER_ROUTING_KEY = "x-dead-letter-routing-key";

    // 郵件隊列
    private String FANOUT_EMAIL_QUEUE = "fanout_email_queue";

    // 短信隊列
    private String FANOUT_SMS_QUEUE = "fanout_sms_queue";
    // fanout 交換機
    private String EXCHANGE_NAME = "fanoutExchange";

    // 1.定義郵件隊列
    @Bean
    public Queue fanOutEamilQueue() {
        // 將普通隊列綁定到死信隊列交換機上
        Map<String, Object> args = new HashMap<>(2);
        args.put(DEAD_LETTER_QUEUE_KEY, deadExchangeName);
        args.put(DEAD_LETTER_ROUTING_KEY, deadRoutingKey);
        Queue queue = new Queue(FANOUT_EMAIL_QUEUE, true, false, false, args);
        return queue;
    }

    // 2.定義短信隊列
    @Bean
    public Queue fanOutSmsQueue() {
        return new Queue(FANOUT_SMS_QUEUE);
    }

    // 2.定義交換機
    @Bean
    FanoutExchange fanoutExchange() {
        return new FanoutExchange(EXCHANGE_NAME);
    }

    // 3.隊列與交換機綁定郵件隊列
    @Bean
    Binding bindingExchangeEamil(Queue fanOutEamilQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanOutEamilQueue).to(fanoutExchange);
    }

    // 4.隊列與交換機綁定短信隊列
    @Bean
    Binding bindingExchangeSms(Queue fanOutSmsQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanOutSmsQueue).to(fanoutExchange);
    }

    /**
     * 建立配置死信郵件隊列
     * 
     * @return
     */
    @Bean
    public Queue deadQueue() {
        Queue queue = new Queue(deadQueueName, true);
        return queue;
    }
   /*
    * 建立死信交換機
    */
    @Bean
    public DirectExchange deadExchange() {
        return new DirectExchange(deadExchangeName);
    }
   /*
    * 死信隊列與死信交換機綁定
    */
    @Bean
    public Binding bindingDeadExchange(Queue deadQueue, DirectExchange deadExchange) {
        return BindingBuilder.bind(deadQueue).to(deadExchange).with(deadRoutingKey);
    }

}

 

 生產者  timestamp 設置爲0 

package com.itmayiedu.rabbitmq;

import java.util.UUID;

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;

@Component
public class FanoutProducer {
    @Autowired
    private AmqpTemplate amqpTemplate;

    public void send(String queueName) {
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("email", "xx@163.com");
        jsonObject.put("timestamp", 0);
        String jsonString = jsonObject.toJSONString();
        System.out.println("jsonString:" + jsonString);
        // 設置消息惟一id 保證每次重試消息id惟一  
        Message message = MessageBuilder.withBody(jsonString.getBytes())
                .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")
                .setMessageId(UUID.randomUUID() + "").build(); //消息id設置在請求頭裏面 用UUID作全局ID 
        amqpTemplate.convertAndSend(queueName, message);
    }
}

 

 

 此時的消費者：

@RabbitListener(queues = "fanout_email_queue")
    public void process(Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        // 獲取消息Id
        String messageId = message.getMessageProperties().getMessageId();
        String msg = new String(message.getBody(), "UTF-8");
        System.out.println("郵件消費者獲取生產者消息msg:"+msg+",消息id"+messageId);
        
        JSONObject jsonObject = JSONObject.parseObject(msg);
        Integer timestamp = jsonObject.getInteger("timestamp");
        
        try {
            int result  = 1/timestamp;
            System.out.println("result"+result);
            // // 手動ack
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            // 手動簽收
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            //拒絕消費消息（丟失消息） 給死信隊列
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
        
        System.out.println("執行結束....");
    }

 

 異常情況：

 

 

添加死信隊列的消費者，並啓動後：

 

package com.itmayiedu.rabbitmq;

import java.util.Map;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

//死信隊列
@Component
public class FanoutDeadEamilConsumer {    
    
    @RabbitListener(queues = "dead_queue")
    public void process(Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        // 獲取消息Id
        String messageId = message.getMessageProperties().getMessageId();
        String msg = new String(message.getBody(), "UTF-8");
        System.out.println("死信郵件消費者獲取生產者消息msg:"+msg+",消息id"+messageId);
        // // 手動ack
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
       // 手動簽收
       channel.basicAck(deliveryTag, false);
        
        System.out.println("執行結束....");
    }    
    
}