在專案開發設計過程中，我們經常用到非同步推送的場景，比如下單和扣減庫存非同步執行。常用的非同步中介軟體有mq有很多，我們這裡以Rabbitmq為例進行講解。我們知道只要是涉及非同步場景的問題，就有成功失敗之分，如果生產者傳送訊息一直失敗要怎麼做呢，這裡我們就用到了死信佇列來實現失敗訊息的處理。

流程圖如下圖：

廢話不多說直接上程式碼：

rabbitmq配置類：

package com.mq.test.config;


import com.rabbitmq.client.Address;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import 
 
com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;

/**
 * Created by Administrator on 2018/6/7.
 */
public class RabbitmqConfig {
    private final static String IP = "172.16.11.119";
    private final static int PORT = 5672;
    private final static String USERNAME = "root";
    private final static 
 
String PASSWORD = "root";
    private static Connection connection = null;
    private static ConnectionFactory factory = null;

    static {
        factory = new ConnectionFactory();
        Address[] addresses = new Address[]{new Address(IP, PORT)};
        factory.setUsername(USERNAME);
        factory
 
.setPassword(PASSWORD);
        try {
            connection = factory.newConnection(addresses);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static Channel getChannel() throws IOException {
        if (connection == null) {
            if (factory != null) {
                return factory.newConnection().createChannel();
            }else{
                return null;
            }
        }
        return connection.createChannel();
    }
}

生產者類：

package com.mq.test;

import com.mq.test.config.RabbitmqConfig;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;

/**
 * Created by Administrator on 2018/6/7.
 */
public class MqProducerTest {
    private final static String DLX_EXCHANGE_NAME = "dlx_exchange";
    private final static String NORMAL_EXCHANGE_NAME = "normal_exchange";
//    private final static String DLX_ROUTINGKEY= "dlx";
private final static String NORMAL_ROUTINGKEY= "normal";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitmqConfig.getChannel();
        channel.exchangeDeclare(NORMAL_EXCHANGE_NAME,"direct",true);
        channel.exchangeDeclare(DLX_EXCHANGE_NAME,"direct",true);
        channel.basicPublish(NORMAL_EXCHANGE_NAME,NORMAL_ROUTINGKEY, MessageProperties.PERSISTENT_TEXT_PLAIN,"dlx test...".getBytes());
    }
}
消費者類：

package com.mq.test;

import com.mq.test.config.RabbitmqConfig;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by Administrator on 2018/6/7.
 */
public class MqConsumerTest {
    private final static String DLX_EXCHANGE_NAME = "dlx_exchange";
    private final static String NORMAL_EXCHANGE_NAME = "normal_exchange";
    private final static String DLX_ROUTINGKEY = "dlx";
    private final static String NORMAL_ROUTINGKEY = "normal";
    private final static String DLX_QUEUE = "dlx";
    private final static String NORMAL_QUEUE = "normal";

    public static void main(String[] args) throws IOException {
        final Channel channel = RabbitmqConfig.getChannel();
        //建立正常佇列和繫結，帶引數的
Map<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
        arguments.put("x-dead-letter-routing-key", DLX_ROUTINGKEY);
        arguments.put("x-message-ttl",6000);//如果想實現延遲佇列的功能可以設定，只要訊息在佇列達到設定的這個時間沒有被消費，就自動傳送到死信佇列，注意這裡是訊息未被消費，而不是未被ack!!!
        channel.queueDeclare(NORMAL_QUEUE, true, false, false, arguments);
        channel.queueBind(NORMAL_QUEUE, NORMAL_EXCHANGE_NAME, NORMAL_ROUTINGKEY);
        //建立死信佇列和繫結
channel.queueDeclare(DLX_QUEUE, true, false, false, null);
        channel.queueBind(DLX_QUEUE, DLX_EXCHANGE_NAME, DLX_ROUTINGKEY);
        channel.basicQos(64);
        channel.basicConsume(NORMAL_QUEUE, false, "myConsumerTag", new Consumer() {
            @Override
public void handleConsumeOk(String s) {

            }

            @Override
public void handleCancelOk(String s) {

            }

            @Override
public void handleCancel(String s) throws IOException {

            }

            @Override
public void handleShutdownSignal(String s, ShutdownSignalException e) {

            }

            @Override
public void handleRecoverOk(String s) {

            }

            @Override
public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                String routingKey = envelope.getRoutingKey();
                String contentType = basicProperties.getContentType();
                long deliveryTag = envelope.getDeliveryTag();
                try {
                    Thread.sleep(8000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
//                channel.basicReject(deliveryTag,false);
}
        });
    }
}
延時或是死信佇列實現：

package com.mq.test;

import com.mq.test.config.RabbitmqConfig;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created by Administrator on 2018/6/7.
 */
public class MqdeplayQueueTest {
    private final static String DLX_QUEUE = "dlx";
    public static void main(String[] args) throws IOException {
        final Channel channel = RabbitmqConfig.getChannel();
        channel.basicConsume(DLX_QUEUE, true, "myDlxConsumer", new Consumer() {
            @Override
public void handleConsumeOk(String s) {

            }

            @Override
public void handleCancelOk(String s) {

            }

            @Override
public void handleCancel(String s) throws IOException {

            }

            @Override
public void handleShutdownSignal(String s, ShutdownSignalException e) {

            }

            @Override
public void handleRecoverOk(String s) {

            }

            @Override
public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
                long deliveryTag = envelope.getDeliveryTag();
                System.out.println("這是延遲佇列收到的訊息："+new String(bytes));
            }
        });
    }
}

這就是全部的程式碼，歡迎大家交流！