ActiveMQ异步发送（异步投递）

素颜马尾好姑娘i 2023-05-21 08:55 36阅读 0赞

## 同步发送 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]

1.  `ActiveMQ默认是使用异步发送模式`
2.  在发送`非事物持久化消息`的时候默认使用的是`同步发送模式`
3.  同步发送时，`Producer.send() 方法会被阻塞`，直到 broker 发送一个确认消息给生产者，这个确认消息暗示生产者 broker 已经成功地将消息持久化到了磁盘
4.  同步发送持久消息能够提供更好的可靠性，但这潜在地`影响了程序的响应速度`，因为在接受到 broker 的确认消息之前应用程序或线程会被阻塞。如果应用程序能够容忍一些消息的丢失，那么可以使用异步发送。异步发送不会在受到 broker 的确认之前一直阻塞 Producer.send 方法。

## 异步投递 ##

1.  异步投递它可以最大化producer端的发送效率，`我们通常在发送消息量比较密集的情况下使用异步发送`，它可以很大的提示producer性能
2.  异步投递也带来了额外的问题，需要较多的Client端的 内存同时也会导致broker端的性能消耗增加
3.  异步投递它`不能有效的保证确认消息发送成功`。在使用 useAsyncSend=true的情况下 客户端需要容忍消息丢失的可能。
4.  `使用异步发送可以提高系统的性能。在默认大多数情况下，AcitveMQ 是以异步模式发送消息` （但发送非事务的持久话消息是同步的）
    
    > 例外的情况：在没有使用事务的情况下，生产者以PERSISTENT 传送模式发送消息。在这种情况下，send 方法都是同步的，并且一直阻塞直到ActiveMQ 发回确认消息：消息已经存储在持久性数据存储中。这种确认机制保证消息不会丢失，但会造成生产者阻塞从而影响反应时间。高性能的程序一般都能容忍在故障情况下丢失少量数据。如果编写这样的程序，可以通过使用异步发送来提高吞吐量（甚至在使用PERSISTENT 传送模式的情况下）。

### 设置异步投递的方式： ###

1.  设置ConnectionFactory时指定使用异步
    
        cf = new ActiveMQConnectionFactory("tcp://locahost:61616?jms.useAsyncSend=true");
2.  修改ConnectionFactory的配置
    
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory("ACTIVEMQ_URL");
        activeMQConnectionFactory.setUseAsyncSend(true);

### 异步投递丢失消息的场景解决方案 ###

**消息丢失场景：**

生产者设置UseAsyncSend=true，使用producer.send(msg)持续发送消息。由于消息不阻塞，生产者会认为所有send的消息均被成功发送至MQ。如果`服务端突然宕机，此时生产者端内存中尚未被发送至MQ的消息都会丢失`

**异步投递丢失消息的解决方案**

1.  正确的异步发送方法是需要接收回调的。
2.  producer.send有带上AsyncCallback的方法。该方法中需要重写onSuccess方法和onException方法。onSuccess方法就是表示这条消息成功发送到MQ上，并接收到了MQ持久化后的回调

代码如下：注意`send方法里面的回调`

public class ActivemqProducer { 
        static String BROCKER_URL = "tcp://192.168.72.129:61616";
        static String QUEUE_NAME = "queue_name8";
    
        public static void main(String[] args) throws JMSException { 
            // 1. 创建连接工厂
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(BROCKER_URL);
            // 设置异步投递方式
            activeMQConnectionFactory.setUseAsyncSend(true);
            // 2. 获取连接 并启动
            Connection connection = activeMQConnectionFactory.createConnection();
            connection.start();
            // 3. 创建session ,参数一：事务，参数二：签收
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            //4. 创建目的地,队列
            Queue queue = session.createQueue(QUEUE_NAME);
            // 5.创建消息生产者 类型为ActiveMQMessageProducer
            ActiveMQMessageProducer messageProducer = (ActiveMQMessageProducer) session.createProducer(queue);
            messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
            // 6.通过 messageProducer 发送3条消息
            for (int i = 1; i <= 3; i++) { 
                TextMessage textMessage = session.createTextMessage("消息 ->" + i);
                // 发送消息，异步回调
                messageProducer.send(textMessage, new AsyncCallback() { 
                    public void onSuccess() { 
                        System.out.println("消息确认发送成功。。。");
                    }
    
                    public void onException(JMSException e) { 
                        System.out.println("消息发送失败。。。");
                    }
                });
    
            }
    
            // 7.关闭资源
            messageProducer.close();
            session.close();
            connection.close();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Z5ajEzOTI1NDc1OTU3_size_16_color_FFFFFF_t_70]: /images/20230521/fff1d525b3a3466ab71c0d4a2d184bb6.png