Início / user-28015381

Daniel Nascimento's questions

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

Obrigado pela ajuda!!! Sou novo no Artemis, mas fiquei realmente surpreso com seu potencial!

Estou trabalhando com o ActiveMQ Artemis e um cliente Python criado com o Qpid Proton (versão 0.39.0). Estou tentando entender como rejeitar ou liberar uma mensagem no Proton Python para que o Artemis tente a reentrega , respeitando a <max-delivery-attempts>configuração definida em broker.xml.

Aqui está o contexto:

  • Versão do Artemis: 2.40.0 (Docker)
  • Versão do Qpid Proton Python: 0.39.0
  • Cliente Python: Consome mensagens usandoproton.handlers.MessagingHandler
  • Comportamento esperado: Se o cliente não conseguir processar uma mensagem (por exemplo, erro interno), ele deverá sinalizar ao Artemis para tentar novamente a entrega. Após o número configurado de tentativas ( max-delivery-attempts), o Artemis deverá mover a mensagem para o DLQ .

seção relevante do 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>

Lógica do cliente Python:

Em on_message, liberamos a mensagem quando o retorno de chamada HTTP falha:

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

Também tentamos:

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

Mas nenhuma das duas pareceu fazer Artemis aumentar a contagem de entregas . Artemis continua entregando a mesma mensagem infinitamente e nunca a move para o DLQ.

Também tentamos adicionar isso (sugerido na documentação do Artemis), mas o Artemis não conseguiu iniciar:

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

Pergunta:

Qual é a maneira correta de liberar ou rejeitar uma mensagem no Qpid Proton Python para que Artemis:

  1. Considera isso uma tentativa de entrega falhada e
  2. Eventualmente envia para o DLQ depois max-delivery-attempts?

Muito obrigado novamente!!

RESOLUÇÃO: Eu não estava atualizando o objeto de estado local antes de liberar a mensagem. Portanto, para usar a política max-delivery-attempts com o DLQ definido em broker.xml, é necessário preencher estas propriedades locais:

def on_message(self, evento): mensagem = evento.mensagem status = send_message_callback(self.inscrição["url_de_destino"], mensagem.corpo)

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

Estou usando o Apache ActiveMQ Artemis 2.40.0 como broker e o Qpid Proton Python 0.39.0 ( python-qpid-proton) como cliente AMQP 1.0.

O objetivo é que as mensagens sejam reenviadas até um limite configurado ( max-delivery-attempts) quando o consumidor explicitamente não reconhece (NACK) uma mensagem, usando o release(delivered=True)método do Proton MessagingHandler.

Configurei o broker.xmlno Artemis com o seguinte <address-setting>bloco:

<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>

O código Python:

Aqui está um trecho do on_messagemétodo no meu consumidor 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)

Apesar de usar release(delivered=True), o Artemis parece reenviar a mensagem indefinidamente (ignorando o <max-delivery-attempts>), a menos que eu mude para reject(event.delivery), que envia imediatamente a mensagem para o DLQ na primeira falha.

Também tentei adicionar a tag <persist-delivery-count-before-delivery>true</persist-delivery-count-before-delivery>ao broker.xml, mas sem sucesso.

Comportamento esperado:

  1. Em seguida release(delivered=True), Ártemis deve:

    • Contar uma tentativa de reentrega
    • Tente novamente a mensagem com base em max-delivery-attemptseredelivery-delay
    • Envie para o DLQ após 5 tentativas frustradas

  2. release(delivered=True)deve se comportar como um NACK suave , conforme descrito na documentação do Proton

Problema:

Artemis continua reenviando a mensagem infinitamente , nunca contando para max-delivery-attempts. Parece que a contagem de entrega não está sendo persistida ou reconhecida como esperado do Proton.

O que eu tentei:

  • Explícitorelease(delivery, delivered=True)
  • Usando reject()(funciona, mas não é o comportamento que eu quero)
  • Adicionando <persist-delivery-count-before-delivery>à configuração do broker (quebra o esquema no Artemis 2.40.0)
  • Verificou-se que os Protons on_released, on_settled, não on_rejectedsão acionados emrelease(delivered=True)

Pergunta:

  • É release(delivered=True)suportado pelo Artemis 2.40.0 como uma forma de incrementar a contagem de entregas?
  • Existe uma maneira correta/melhor de fazer NACK no Proton Python para que o Artemis rastreie as reentregas?
  • Existe um broker alternativo ou configuração do Proton para dar suporte a essa lógica de repetição?

Desde já, obrigado!

RESOLUÇÃO:

Eu não estava atualizando o objeto de estado local antes de liberar a mensagem. Então, para usar a max-delivery-attemptspolítica com o DLQdefinido em broker.xml, ele deve completar estas propriedades locais:

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