ActiveMQ

高级特性

1 异步投递

(1) 异步投递是什么

自我理解：异步投递，是指生产者和broker之间发送消息的异步。不是指生产者和消费者之间异步。

特征：

① 不能有效保证消息的发送成功

② 如果出现slowConsumer，可能会给Broker带来消息积压的可能

官网介绍：http://activemq.apache.org/async-sends

说明：对于一个Slow Consumer,使用同步发送消息可能出成Producer堵塞等情况，慢消费者适合使用异步发送。

(这句话我认为有误)

总结：

①　异步发送可以让生产者发的更快。

扫描二维码关注公众号，回复： 12647488 查看本文章

②　如果异步投递不需要保证消息是否发送成功，发送者的效率会有所提高。如果异步投递还需要保证消息是否成功发送，并采用了回调的方式，发送者的效率提高不多，这种就有些鸡肋。

(2) 代码实现

官网上3中代码实现：

public class Jms_TX_Producer {
    
    
    // 方式1。3种方式任选一种
    private static final String ACTIVEMQ_URL = "tcp://118.24.20.3:61626?jms.useAsyncSend=true";//设置异步投递
    private static final String ACTIVEMQ_QUEUE_NAME = "Async";

    public static void main(String[] args) throws JMSException {
    
    
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        // 方式2
        activeMQConnectionFactory.setUseAsyncSend(true);//设置异步投递
        Connection connection = activeMQConnectionFactory.createConnection();
        // 方式3
        ((ActiveMQConnection)connection).setUseAsyncSend(true);//设置异步投递
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);
        MessageProducer producer = session.createProducer(queue);
        try {
    
    
            for (int i = 0; i < 3; i++) {
    
    
                TextMessage textMessage = session.createTextMessage("tx msg--" + i);
                producer.send(textMessage);
            }
            System.out.println("消息发送完成");
        } catch (Exception e) {
    
    
            e.printStackTrace();
        } finally {
    
    
            producer.close();
            session.close();
            connection.close();
        }
    }
}

(3) 异步发送如何确认发送成功

public class JmsProduce_asynsc {
    
    

    public static final String ACTIVEMQ_URL = "tcp://192.168.109.101:61616";
    public static final String QUEUE_NAME = "asynsc01";

    public static void main(String[] args) throws JMSException {
    
    
        ActiveMQConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        mqConnectionFactory.setUseAsyncSend(true);//设置异步投递
        Connection conn = mqConnectionFactory.createConnection();
        conn.start();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(QUEUE_NAME);
        ActiveMQMessageProducer mqMessageProducer = (ActiveMQMessageProducer) session.createProducer(queue);
        mqMessageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
        TextMessage textMessage = null;
        for (int i = 1; i <= 3; i++) {
    
    
            textMessage = session.createTextMessage("msg--asynsc01-----" + i);
            textMessage.setJMSMessageID(UUID.randomUUID().toString().substring(0,3)+"---orderAchang");
            String msgID = textMessage.getJMSMessageID();
            //new AsyncCallback()设置异步投送的回调函数
            mqMessageProducer.send(textMessage, new AsyncCallback() {
    
    
                @Override
                public void onSuccess() {
    
    
                    //发送成功的情况
                    System.out.println(msgID+"has benn ok send");
                }

                @Override
                public void onException(JMSException exception) {
    
    
                    //发送失败的情况
                    System.out.println(msgID+"fail to send to mq");
                }
            });

        }
        mqMessageProducer.close();
        session.close();
        conn.close();

        System.out.println("========消息发布到MQ完成===========");
    }

}

控制台观察发送消息的信息：

2 延迟投递和定时投递

(1) 介绍

官网文档：http://activemq.apache.org/delay-and-schedule-message-delivery.html

(2) 修改配置文件并重启

 </bean>
    <broker xmlns="http://activemq.apache.org/schema/core" brokerName="localhost" dataDirectory="${activemq.data}"  schedulerSupport="true" >
        <destinationPolicy>

之后重启activemq

(3) 代码实现

生产者代码：

public class JmsProduce_delayAndschedule {
    
    

    public static final String ACTIVEMQ_URL = "tcp://192.168.109.101:61616";
    public static final String QUEUE_NAME = "asynsc01";

    public static void main(String[] args) throws JMSException {
    
    
        ActiveMQConnectionFactory mqConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        mqConnectionFactory.setUseAsyncSend(true);//设置异步投递
        Connection conn = mqConnectionFactory.createConnection();
        conn.start();
        Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(QUEUE_NAME);
        ActiveMQMessageProducer mqMessageProducer = (ActiveMQMessageProducer) session.createProducer(queue);
==================================================================================      
        long delay = 3 * 1000; //延迟3秒
        long period = 4 * 1000; //四秒钟投递一次
        int repeat = 5; //投递次数

        TextMessage textMessage = null;
        for (int i = 1; i <= 3; i++) {
    
    
            textMessage = session.createTextMessage("msg--delay--正文内容-----" + i);

            //消息属性设置
            textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY,delay);
            textMessage.setLongProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD,period);
            textMessage.setIntProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT,repeat);
            // 此处的意思：该条消息，等待3秒，之后每4秒发送一次，重复发送5次。
=====================================================================================
            mqMessageProducer.send(textMessage);

        }
        mqMessageProducer.close();
        session.close();
        conn.close();

        System.out.println("========消息发布到MQ完成===========");
    }

}

消费者代码：与之前一样

3 消息消费的重试机制

(1) 是什么

官网文档：http://activemq.apache.org/redelivery-policy

是什么： 消费者收到消息，之后出现异常了，没有告诉broker确认收到该消息，broker会尝试再将该消息发送给消费者。尝试n次，如果消费者还是没有确认收到该消息，那么该消息将被放到死信队列重，之后broker不会再将该消息发送给消费者。

(2) 具体哪些情况会引发消息重发

①　Client用了transactions且再session中调用了rollback

②　Client用了transactions且再调用commit之前关闭或者没有commit

③　Client再CLIENT_ACKNOWLEDGE的传递模式下，session中调用了recover

(3) 请说说消息重发时间间隔和重发次数

间隔：1

次数：6

每秒发6次

(4) 有毒消息Poison ACK

一个消息被redelivedred超过默认的最大重发次数（默认6次）时，消费的回个MQ发一个“poison ack”表示这个消息有毒，告诉broker不要再发了。这个时候broker会把这个消息放到DLQ（死信队列）。

(5) 属性说明

(6） 代码验证

生产者。发送3条数据。代码省略…

消费者。开启事务，却没有commit。重启消费者，前6次都能收到消息，到第7次，不会再收到消息。代码：

public class Jms_TX_Consumer {
    
    
    private static final String ACTIVEMQ_URL = "tcp://192.168.109.101:61616";
    private static final String ACTIVEMQ_QUEUE_NAME = "dead01";

    public static void main(String[] args) throws JMSException, IOException {
    
    
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        final Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);
        MessageConsumer messageConsumer = session.createConsumer(queue);
        messageConsumer.setMessageListener(new MessageListener() {
    
    
            public void onMessage(Message message) {
    
    
                if (message instanceof TextMessage) {
    
    
                    TextMessage textMessage = (TextMessage) message;
                    try {
    
    
                        System.out.println("***消费者接收到的消息:   " + textMessage.getText());
                        //session.commit();
                    }catch (Exception e){
    
    
                        e.printStackTrace();
                    }
                }
            }
        });
        //关闭资源
        System.in.read();
        messageConsumer.close();
        session.close();
        connection.close();
    }
}

activemq管理后台。多了一个名为ActiveMQ.DLQ队列，里面多了3条消息。

(7) 代码修改默认参数

修改重试次数为3。更多的设置请参考官网文档。

消费者代码：

public class Jms_TX_Consumer {
    
    
    private static final String ACTIVEMQ_URL = "tcp://118.24.20.3:61626";
    private static final String ACTIVEMQ_QUEUE_NAME = "dead01";

    public static void main(String[] args) throws JMSException, IOException {
    
    
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ACTIVEMQ_URL);
=====================================================================
        // 修改默认参数，设置消息消费重试3次
        RedeliveryPolicy redeliveryPolicy = new RedeliveryPolicy();
        redeliveryPolicy.setMaximumRedeliveries(3);
        activeMQConnectionFactory.setRedeliveryPolicy(redeliveryPolicy);
=====================================================================
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        final Session session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
        Queue queue = session.createQueue(ACTIVEMQ_QUEUE_NAME);
        MessageConsumer messageConsumer = session.createConsumer(queue);
        messageConsumer.setMessageListener(new MessageListener() {
    
    
            public void onMessage(Message message) {
    
    
                if (message instanceof TextMessage) {
    
    
                    TextMessage textMessage = (TextMessage) message;
                    try {
    
    
                        System.out.println("***消费者接收到的消息:   " + textMessage.getText());
                        //session.commit();
                    }catch (Exception e){
    
    
                        e.printStackTrace();
                    }
                }
            }
        });
        System.in.read();
        messageConsumer.close();
        session.close();
        connection.close();
    }
}

(8)整合spring

4 死信队列

承接上个标题的内容。

(1) 是什么

官网文档： http://activemq.apache.org/redelivery-policy

死信队列：异常消息规避处理的集合，主要处理失败的消息。

(1) 死信队列的配置（一般采用默认）

1. sharedDeadLetterStrategy

不管是queue还是topic，失败的消息都放到这个队列中。下面修改activemq.xml的配置，可以达到修改队列的名字。

2. individualDeadLetterStrategy

可以为queue和topic单独指定两个死信队列。还可以为某个话题，单独指定一个死信队列。

3. 自动删除过期消息

过期消息是值生产者指定的过期时间，超过这个时间的消息。

4. 存放非持久消息到死信队列中

5 消息不被重复消费，幂等性

如何保证消息不被重复消费呢？幕等性问题你谈谈

也就是变向问：如何解决消息的重复消费问题？

幂等性如何解决，根据messageid去查这个消息是否被消费了

感谢尚硅谷