Поддерживает ли реализация python activemq поддержку бессрочной подписки?

Question

Мне нужен stomp.py потребитель ActiveMQ в качестве процесса-демона для прослушивания определенной очереди. В настоящее время я сделал обходной путь, который подпишется на очередь, затем time.sleep(60) секунд и отписаться в цикле WHILE TRUE. Таким образом, я могу снова и снова подписываться на очередь. Но проблема в том, что каждому сообщению требуется 13 секунд для выполнения каждого сообщения N / 5 (N = общее количество сообщений в очереди), которое дублируется. Потому что во время обработки (N / 5)-го сообщения ограничение по времени пересекает (60) -ю секунду. тогда отписались, позвонили. без подтверждения снова обрабатывается то же сообщение.

Способ просмотра:

1. on_message: происходит обработка сообщения.
2. 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()

Метод подписки с неопределенным временем.