死信队列

默认情况下，消息六次消费失败后，会将该消息移到ActiveMQ.DLQ队列中

生产数据端
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.qpid.jms.JmsConnectionFactory;

import javax.jms.*;

public class Producer {

    public static void main(String[] args) {
        String protocol = "tcp://120.25.242.46:61616";
        new ProducerThread(protocol, "Dead-Letter").start();
    }

    static class ProducerThread extends Thread {
        String brokerUrl;
        String destinationUrl;
        public ProducerThread(String brokerUrl, String destinationUrl) {
            this.brokerUrl = brokerUrl;
            this.destinationUrl = destinationUrl;
        }

        public void run() {
            ConnectionFactory connectionFactory;
            Connection conn;
            Session session;
            try {
                // 1 创建连接工厂
                connectionFactory = new ActiveMQConnectionFactory(null, null, brokerUrl);
                // 2 创建连接
                conn = connectionFactory.createConnection();
                conn.start();
                // 3 创建会话
                session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
                // 4 创建消息发送目标
                Destination destination = session.createQueue(destinationUrl);
                // 5 用亩的地创建消息生产者
                MessageProducer producer = session.createProducer(destination);
                // 6 设置递送模式
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);
                producer.setPriority(7);
                // 7 通过producer 发送消息
                TextMessage textMessage = session.createTextMessage("11111111");
                producer.send(textMessage);
                session.close();
                conn.close();
            } catch(Exception e) {
                e.printStackTrace();
            }
        }
    }
}

消费数据端
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.qpid.jms.JmsConnectionFactory;

import javax.jms.*;

/**
 * 简单消费者
 */
// http://activemq.apache.org/consumer-features.html
public class Consumer {
    public static void main(String[] args) {
        new ConsumerThread("tcp://120.25.242.46:61616", "Dead-Letter").start();
    }
}

class ConsumerThread extends Thread {

    String brokerUrl;
    String destinationUrl;

    public ConsumerThread(String brokerUrl, String destinationUrl) {
        this.brokerUrl = brokerUrl;
        this.destinationUrl = destinationUrl;
    }

    @Override
    public void run() {
        ConnectionFactory connectionFactory;
        Connection conn;
        Session session;
        MessageConsumer consumer;

        try {
            // brokerURL
            // http://activemq.apache.org/connection-configuration-uri.html
            // 1、创建连接工厂
            connectionFactory = new ActiveMQConnectionFactory(null, null, this.brokerUrl);

            // 2、创建连接对象
            conn = connectionFactory.createConnection();
            conn.start(); // 一定要启动

            // 3、创建会话（可以创建一个或者多个session）
            session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 4、创建消息消费目标(Topic or Queue)
            Destination destination = session.createQueue(destinationUrl);

            // 5、创建消息消费者 http://activemq.apache.org/destination-options.html
            consumer = session.createConsumer(destination);

            // 6、异步接收消息
            consumer.setMessageListener(new MessageListener() {

                @Override
                public void onMessage(Message message) {
                    // 抛出异常后，不会消费该消息。重新七次后，会将该消息移到ActiveMQ.DLQ队列中。
                    int i = 1/0;
                }
            });
            try {
                // 担心收不到消息就关闭了，先睡眠一秒
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            consumer.close();
             session.close();
             conn.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}