Тайм-аут обработки сообщений на уровне потребителя в activemq - PullRequest
Новая
       31

Тайм-аут обработки сообщений на уровне потребителя в activemq

0 голосов
/ 16 мая 2019

У меня потребитель подписывается на очередь, используя метод подписки с режимом ack = client-personal. когда происходит исключение, оно зависает, а не обрабатывает следующее сообщение. В блоке try у меня есть метод connection.ack (), а в блоке catch я просто записал исключение, потому что отправка connection.nack () отправляет сообщение в очередь недоставленных сообщений.

1). Я попытался сделать NACK, который отправляет сообщение в DLQ (не желательно). 2). Я создал длинного пользователя для опроса, который отписывается через определенное время и снова подписывается таким образом, что я получаю одно и то же сообщение.

образец потребителя:

Примечание:

Рассмотрим исключение при вызове func_to_exec в методе on_message.

методов для обзора:

on_message: где изящно обрабатывается последний блок исключений. потребление_msg: где происходит подписка и отмена подписки, в то время как ИСТИНА 

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, то это же сообщение было доставлено (или) потребителю, способному выбрать другое сообщение после истечения времени ожидания обработки сообщения.

1 Ответ

0 голосов
/ 16 мая 2019

Нет ActiveMQ не имеет такой функции, вам нужно либо NACK сообщение или отписаться, чтобы разрешить доставку полученных сообщений другому потребителю.

...