tuoheng_airport_stream_discard_20240715/concurrency/mqtt/MqttFeedbackThread.py

# -*- coding: utf-8 -*-
import time
from threading import Thread
from traceback import format_exc

from loguru import logger

from util.QueUtil import get_block_queue

'''
    问题反馈线程
'''


class FeedbackThread(Thread):
    __slots__ = ('__fb_queue', '__mq')

    def __init__(self, fb_queue, mq):
        super().__init__()
        self.__fb_queue = fb_queue
        self.__mq = mq

    def run(self):
        logger.info("启动反馈线程")
        fb_queue, mq = self.__fb_queue, self.__mq
        while True:
            try:
                fb = get_block_queue(fb_queue)
                if fb is not None and len(fb) > 0:
                    mq.publish(fb)
                else:
                    time.sleep(1)
            except Exception:
                logger.error("反馈异常：{}", format_exc())
        logger.info("反馈线程执行完成")