python - activemq 中的消费者级别消息处理超时

问题描述

我有一个消费者使用 ack=client-individual 模式的 subscribe 方法订阅队列。当发生异常时，它会挂断而不是处理下一条消息。在 try 块中，我有 connection.ack() 方法，而在 catch 块中，我刚刚记录了异常，因为发送 connection.nack() 会将消息发送到死信队列。

1）。我试图做一个 NACK 向 DLQ 发送消息（不需要）。2）。我创建了一个长轮询消费者，它在特定时间后取消订阅并以这种方式再次订阅，我得到了相同的消息。

样本消费者：

笔记：

考虑在 on_message 方法中的 func_to_exec 调用中发生异常。

审查方法：

on_message ：优雅地处理最终异常块的位置。consume_msg ：订阅和取消订阅发生在当 TRUE 时

import sys
import stomp
import time
import json
from core.lib.foundation.activemq_wrappers.activemq_configmanager import MQConfigMgnr

ACK_MODES = ("auto", "client", "client-individual")

class CustomListener(stomp.ConnectionListener):

    def __init__(self,conn, func_to_exec,ack_mode):
        self.conn = conn
        self.func_to_exec = func_to_exec
        self.ack_mode = ack_mode

    def on_error(self, headers, message):
        print('received an error::%s' % message)

    def call_nack(self, msg_id, subscription):
        try:
            if self.ack_mode!=0:
                print("nack done..!")
                self.conn.nack(msg_id, int(subscription))
        except Exception as ex:
            raise ex

    def call_ack(self, msg_id, subscription):
        try:
            if self.ack_mode!=0:
                print("ack done..!")
                self.conn.ack(msg_id, int(subscription))
        except Exception as ex:
            raise ex

    def on_message(self, headers, message):
        try:

            try:
                message = json.loads(message)
            except Exception:
                self.call_nack(headers.get("message-id"),headers.get("subscription"))
            else:
                response = self.func_to_exec(message,headers)
                self.call_ack(headers.get("message-id"),headers.get("subscription"))
                if 'reply-to' in headers:
                    self.conn.send(destination='/queue/{}'.format(headers.get("reply-to")),
                                   body=json.dumps(response), header={"JMSDeliveryMode":"Persistent",
                                                                      "JMSPriority":4})
        except Exception as ex:
            print("exception in on_message::%s"%ex)

    def on_disconnected(self):
        print('disconnected')

class QueueConsumer:

    def __init__(self, section_name, ack_mode=2):
        try:
            self.ack_mode = ack_mode
            self.configobj = MQConfigMgnr()
            self.section = self.configobj.get_config("ACTIVEMQ_PARAMS")
            self.hosts = self.section.get("hosts")
            self.ports = self.section.get("ports")
            self.ENCODE_FORMAT = self.section.get("ENCODE_FORMAT")
            self.conn_param = zip(self.hosts,self.ports)
            self.conn = stomp.Connection11(self.conn_param, encoding=self.ENCODE_FORMAT)
            func_to_exec = self.get_execution_func(section_name)
            self.conn.start()
            self.conn.connect(wait=True,)
            self.conn.set_listener('', CustomListener(self.conn, func_to_exec,ack_mode))
        except stomp.exception.ConnectFailedException as stomp_ex:
            raise stomp_ex
        except Exception as ex:
            raise ex

    def get_execution_func(self,section_name):
        try:
            consumer_data = self.configobj.get_config("CONSUMERS")["QUEUES"]
            file_name = consumer_data[section_name]["FILE_NAME"]
            func_to_exec = consumer_data[section_name]["FUNC_TO_EXEC"]
            self.queue_names = consumer_data[section_name]["QUEUE_NAME"]
            self.queue_names = ",".join(map((lambda x:"/queue/{}".format(x)),self.queue_names))
            print(self.queue_names)
            mod = __import__(file_name, fromlist=['*'])
            func_to_execute = mod.__dict__[func_to_exec]
            return func_to_execute
        except Exception as ex:
            raise ex

    def consume_msg(self):
        try:
            while True:
                self.conn.subscribe(self.queue_names,1, ack=ACK_MODES[self.ack_mode],
                                headers={"activemq.prefetchSize":1}
                                )
                time.sleep(60)
                self.conn.unsubscribe(1)
        except Exception as ex:
            raise ex

    def close_connection(self):
        self.conn.disconnect()

    def execute(self):
        try:
            self.consume_msg()
            self.close_connection()
        except Exception as ex:
            raise ex

if __name__ == "__main__":
    try:
        section_name = sys.argv[1]
    except IndexError as ex:
        print("please provide section name")
        sys.exit(0)
    obj = QueueConsumer(section_name,ack_mode=2)
    obj.execute()

我们能否为消费者中的消息设置最大消息处理时间，以便如果消息处理达到 MAX_TIME 则相同的消息将被重新传递（或）消费者能够在消息处理超时后选择另一条消息。

标签： pythonactivemqproducer-consumer

python - activemq 中的消费者级别消息处理超时

问题描述

样本消费者：

笔记：

审查方法：

解决方案

推荐阅读