主页 / user-28015381

Daniel Nascimento's questions

Martin Hope
Daniel Nascimento
Asked: 2025-04-06 22:38:34 +0800 CST
  • 5

谢谢你的帮助！！！我是 Artemis 的新手，但它的潜力真的让我很惊讶！

我正在使用ActiveMQ Artemis和一个使用Qpid Proton (版本)构建的 Python 客户端0.39.0。我试图了解如何在 Proton Python 中拒绝或释放一条消息，以便 Artemis尝试重新发送，并遵循<max-delivery-attempts>中配置的设置broker.xml。

以下是上下文：

  • Artemis 版本： 2.40.0（Docker）
  • Qpid Proton Python版本： 0.39.0
  • Python 客户端：使用以下方式消费消息proton.handlers.MessagingHandler
  • 预期行为：如果客户端无法处理消息（例如内部错误），客户端应向 Artemis 发出信号，要求其重试投递。在达到配置的重试次数 ( max-delivery-attempts) 后，Artemis 应将消息移至DLQ。

broker.xml相关部分：

<address-setting match="#">
   <max-delivery-attempts>5</max-delivery-attempts>
   <redelivery-delay>5000</redelivery-delay>
   <dead-letter-address>DLQ</dead-letter-address>
   <auto-create-dead-letter-resources>true</auto-create-dead-letter-resources>
   <dead-letter-queue-suffix>.DLQ</dead-letter-queue-suffix>
</address-setting>

Python客户端逻辑：

在中on_message，我们在HTTP回调失败时发布消息：

self.release(event.delivery, delivered=True)

我们还尝试了：

self.settle(event.delivery, state=Delivery.RELEASED)

但这两种方法似乎都没能让 Artemis增加投递计数。Artemis 一直无限地投递同一条消息，并且永远不会将其移动到 DLQ。

我们也尝试添加这个（在 Artemis 文档中建议），但 Artemis 无法启动：

<persist-delivery-count-before-delivery>true</persist-delivery-count-before-delivery>

问题：

在Qpid Proton Python中发布或拒绝消息的正确方法是什么，以便 Artemis：

  1. 将其视为一次失败的递送尝试，并且
  2. 最终将其发送到DLQ后max-delivery-attempts？

再次感谢！！

解决方案：我在发布消息之前没有更新本地状态对象。因此，为了将 max-delivery-attempts 策略与 broker.xml 中定义的 DLQ 结合使用，它必须完成以下本地属性：

def on_message（self，event）：message = event.message status = send_message_callback（self.enrollment [“target_url”]，message.body）

if 200 <= status < 300:
    self.accept(event.delivery)
else:
    # Explicitly NACK the message
    local_state = event.delivery.local
    local_state.failed = True
    local_state.undeliverable = False
    local_state.type = event.delivery.MODIFIED
    event.delivery.update(local_state)
    self.release(event.delivery, delivered=True)
python
Martin Hope
Daniel Nascimento
Asked: 2025-04-05 21:36:48 +0800 CST
  • 5

我使用Apache ActiveMQ Artemis 2.40.0作为代理，使用Qpid Proton Python 0.39.0 ( python-qpid-proton) 作为 AMQP 1.0 客户端。

目标是max-delivery-attempts当消费者明确不确认(NACK) 消息时，使用release(delivered=True)Proton 的方法重新传递消息，直至达到配置的限制 () MessagingHandler。

我broker.xml在 Artemis 中配置了以下<address-setting>块：

<address-setting match="#">
    <expiry-address>ExpiryQueue</expiry-address>
    <dead-letter-address>DLQ</dead-letter-address>
    <auto-create-dead-letter-resources>true</auto-create-dead-letter-resources>
    <dead-letter-queue-prefix></dead-letter-queue-prefix>
    <dead-letter-queue-suffix>.DLQ</dead-letter-queue-suffix>
    <max-delivery-attempts>5</max-delivery-attempts>
    <redelivery-delay>5000</redelivery-delay>
</address-setting>

Python代码：

on_message以下是我的 Qpid Proton 消费者中方法的片段：

def on_message(self, event):
    message = event.message
    status = send_message_callback(self.enrollment["target_url"], message.body)

    if 200 <= status < 300:
        self.accept(event.delivery)
    else:
        # Explicitly NACK the message
        self.release(event.delivery, delivered=True)

尽管使用，Artemis 似乎会无限期地release(delivered=True)重新传递消息（忽略），除非我切换到，它会在第一次故障时立即将消息发送到 DLQ。<max-delivery-attempts>reject(event.delivery)

我也尝试过将标签添加<persist-delivery-count-before-delivery>true</persist-delivery-count-before-delivery>到broker.xml，但没有成功。

预期行为：

  1. 之后release(delivered=True)，Artemis 应该：

    • 计算一次重新投递尝试
    • max-delivery-attempts根据和重试消息redelivery-delay
    • 5 次尝试失败后将其发送到 DLQ

  2. release(delivered=True)应表现为软 NACK，如 Proton文档中所述

问题：

Artemis不断重复发送消息，从不计入max-delivery-attempts。看来 Proton 并未像预期的那样持久化或确认发送次数。

我尝试过的：

  • 显式release(delivery, delivered=True)
  • 使用reject()（有效，但不是我想要的行为）
  • 添加<persist-delivery-count-before-delivery>到代理配置（破坏 Artemis 2.40.0 中的模式）
  • 已验证 Proton 的on_released、on_settled、on_rejected未在release(delivered=True)

问题：

  • Artemis 2.40.0是否release(delivered=True)支持增加交付次数？
  • 在 Proton Python 中是否有正确/更好的方法来 NACK，以便 Artemis 跟踪重新交付？
  • 是否有替代代理或 Proton 配置来支持此重试逻辑？

提前致谢！

解决：

在发布消息之前，我没有更新本地状态对象。因此，为了使用在 broker.xml 中定义的max-delivery-attempts策略，它必须完成以下本地属性：DLQ

def on_message(self, event):
    message = event.message
    status = send_message_callback(self.enrollment["target_url"], message.body)

    if 200 <= status < 300:
        self.accept(event.delivery)
    else:
        # Explicitly NACK the message
        local_state = event.delivery.local
        local_state.failed = True
        local_state.undeliverable = False
        event.delivery.update(local_state.type)
        self.settle(event.delivery, event.delivery.MODIFIED)
amqp