activeMQ高级特性之异步投递与确认签收回调

异步投递与确认签收回调

• ActiveMQ 支持同步、异步两种发送的模式将消息发送到 broker，模式的选择对发送延时有巨大的影响。producer 能达到怎样的产出率（产出率 = 发送数据总量/时间）主要受发送延时的影响，使用异步发送可以显著的提高发送的性能。


•     ActiveMQ 默认使用异步发送的模式：除非明确指定使用同步发送的方式或者在未使用事务的前提下发送持久化的消息，这两种情况都是同步发送的。


•     如果你没有使用事务且发送的是持久化的消息，每一次发送都是同步发送的且会阻塞 producer 直到 broker 返回一个确认，表示消息己经被安全的持久化到磁盘。确认机制提供了消息安全的保障，但同时会阻塞客户端带来了很大的延时。


•     很多高性能的应用，允许在失败的情况下有少量的数据丢失。如果你的应用满足这个特点，你可以使用异步发送来提高生产率，即使发送的是持久化的消息。

    异步投递可以最大化 produer 端的发送效率。我们通常在发送消息量比较密集的情况下使用异步发送，它可以很大的提升 Producer 性能；

    不过这也带来了额外的问题，就是需要消耗较多的 Client 端内存同时也会导致 broker 端性能消耗增加；此外它不能有效的确保消息的发送成功。

  在 useAsyncSend=true 的情况下客户端需要容忍消息丢失的可能。

异步发送如何确认发送成功？

   异步发送丢失消息的场景是：生产者设置 UseAsyncSend=true，使用 producer.send（msg）发送消息

    由于消息不阻塞，生产者会认为所有 send 的消息均被成功发送至 MQ。如果 MQ 突然宕机，此时生产者端内存中尚未被发送至 MQ 的消息都会丢失。

    所以，正确的异步方法是需要接收回调的

    同步发送和异步发送的区别就在此，同步发送等send不阻塞了就表示一定发送成功了，异步发送需要接收回执并由客户端再判断一次是否发送成功。

非SpringBoot项目中

//使用连接URI配置异步发送
cf = new ActiveMQConnectionFactory("tcp://locahost:61616?jms.useAsyncSend=true");

//在ConnectionFactory 级别配置异步发送
((ActiveMQConnectionFactory)connectionFactory).setUseAsyncSend(true);
//在连接级别配置异步发送
((ActiveMQConnection)connection).setUseAsyncSend(true);


1)SpringBoot项目中：

@Component
public class QueueProducer {
@Autowired
private JmsMessagingTemplate template;
@Autowired
private Queue queue;
@Autowired
private ActiveMQConnection connection;
/**
* 消息同步发送: 在消息发送成功到MQ中之前,Producer端一直处于阻塞状态,消息的发送成功与否可从阻塞为阻塞看出
* 　　　　　　　　1.强制要求同步发送 2.事务持久化发送时
* 消息异步发送: 消息发出后,MQ中未收到消息之前宕机了,但Producer没有显示发送成功与否,会认为发送成功
* 　　　　　　　此时,为避免这种情况-->有异步发送的回调机制,发送成功或失败后会给Producer回调结果
*/
public void produceMessage() {
//template.getConnectionFactory().createConnection().createSession();
/*进行复杂应用设置时,可从template中获取connection,进行原始的操作,但此时使用了pool的session缓存spring.jms.cache.enable默认为true,
　　　　　　返回的Connection为CacheConnectionFactory,无法生成ActiveMQMessageProducer,无法进行send方法的AsyncCallback回调
　　　　　　,所以此时需要创建一个ActiveMQMessageConnection的配置类以支持异步的调用并成功回调*/
try {
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
ActiveMQMessageProducer producer = (ActiveMQMessageProducer) session.createProducer(queue);
ObjectMessage message = session.createObjectMessage();
message.setStringProperty("msgId", "001");
message.setObject(new User(1, "小明"));
producer.send(message, new AsyncCallback() {
@Override
public void onSuccess() {
try {
System.out.println("消息发送成功:" + message.getStringProperty("msgId"));
} catch (JMSException e) {
e.printStackTrace();
}
}
@Override
public void onException(JMSException e) {
try {
System.out.println("出现异常:" + message.getStringProperty("msgId"));
} catch (JMSException je) {
je.printStackTrace();
}
}
});
} catch (JMSException e) {
e.printStackTrace();
}
//template.convertAndSend(queue,"message:"+new Date().toLocaleString());
}

public class ActiveMQConnectionConfig {
@Value("${spring.activemq.broker-url}")
private String brokerUrl;
@Bean
public ActiveMQConnection activeMQConnection() throws JMSException {
ActiveMQConnectionFactory activeMQCOnnectionFactory= new ActiveMQConnectionFactory(brokerUrl);
//启用异步发送消息
activeMQConnectionFactory.setUseAsyncSend(true);
ActiveMQConnection cOnnection=
(ActiveMQConnection) activeMQConnectionFactory.createConnection();
return connection;