ActiveMQ 高级特性

异步投递

异步投递是指生产者与Broker之间的消息发送过程与生产者执行其他任务并行。与传统的同步投递相比，异步投递允许生产者更快速地发送消息，但也伴随着消息发送成功不保证的风险。

异步投递特点

不保证消息成功发送：异步投递无法确保每条消息都被Broker接收，可能会出现消息丢失的情况。

可能导致消息积压：如果消费者（slow consumer）处理消息速度过慢，Broker可能会积压大量消息，影响性能。

异步投递总结

异步投递适合对消息处理速度要求较高的场景，但需要特别注意消息丢失和积压问题。在实际应用中，可以根据需求选择同步或异步投递方式。

延迟投递和定时投递

延迟投递允许生产者在指定时间发送消息，而定时投递则允许生产者在一定周期内重复发送消息。

延迟投递配置

修改配置文件：在activemq.xml中添加以下配置：

重启activemq：完成配置后，重启Broker使其生效。

定时投递实现

生产者可以通过设置消息属性来实现定时投递：

message.setProperty(ScheduledMessage.AMQ_SCHEDULED_DELAY, "3000"); // 延迟3秒message.setProperty(ScheduledMessage.AMQ_SCHEDULED_PERIOD, "4000"); // 每4秒发送一次message.setProperty(ScheduledMessage.AMQ_SCHEDULED_REPEAT, "5"); // 重复5次

消息消费的重试机制

消费者在遇到异常时，Broker会重试发送消息。默认情况下，最多重试6次，每次间隔1秒。

消息重发条件

事务处理异常：在事务中调用rollback。

事务未提交：在事务提交前关闭Session或未 commit。

ACK模式下的Session恢复：在CLIENT_ACKNOWLEDGE模式下调用recover。

消息重发次数和间隔

默认重发次数为6次，间隔为1秒。

有毒消息处理

如果消息重发次数超过默认限制（6次），Broker会将其标记为有毒消息（poison ack），并将其放入死信队列（DLQ）。

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

幂等性是指消息在被消费之前不会被重复消费。可以通过消息ID来确保这一点。

幂等性实现

在消费者代码中，可以检查消息的JMSMessageID，避免重复消费：

if (message instanceof TextMessage) {    TextMessage txtMsg = (TextMessage) message;    String msgId = txtMsg.getJMSMessageID();    // 根据msgId判断是否已消费}

死信队列

死信队列用于处理因消费者无法处理或异常导致的消息。Broker会自动将这些消息路由至死信队列。

死信队列配置

默认死信队列：

单独配置：

过期消息处理：可以通过设置消息的AMQ_SCHEDULED_EXPIRATION属性来指定消息过期时间。

消息幂等性与重复消费

幂等性是确保消息在被消费前不会被重复消费的机制。可以通过消息ID来实现，避免重复处理相同消息。

幂等性实现

在消费者代码中，可以检查消息的JMSMessageID，确保每条消息只被消费一次：

if (session.createConsumer(queue, new MessageListener() {    public void onMessage(Message message) {        if (message instanceof TextMessage) {            TextMessage txtMsg = (TextMessage) message;            String msgId = txtMsg.getJMSMessageID();            // 根据msgId检查是否已消费        }    }});

转载地址：http://nioq.baihongyu.com/

你可能感兴趣的文章