|
- # -*- coding: utf-8 -*-
- import time
- from threading import Thread
- from traceback import format_exc
-
- from loguru import logger
-
- from util.KafkaUtils import CustomerKafkaProducer
-
- '''
- 问题反馈线程
- '''
-
-
- class FeedbackThread(Thread):
- __slots__ = [
- '__fbQueue',
- '__context'
- ]
-
- def __init__(self, fbQueue, context):
- super().__init__()
- self.__fbQueue = fbQueue
- self.__context = context
-
- '''
- 阻塞获取反馈消息
- '''
-
- def getFeedback(self):
- return self.__fbQueue.get()
-
- def run(self):
- logger.info("启动问题反馈线程")
- kafkaProducer = CustomerKafkaProducer(self.__context)
- dsp_push_stream_result_topic = self.__context["kafka"]["topic"]["dsp-push-stream-result-topic"]
- while True:
- logger.info("问题反馈发送消息循环")
- feedback = {}
- try:
- fb = self.getFeedback()
- if fb is not None and len(fb) > 0:
- pull_stream = fb.get("pull_stream")
- if pull_stream is not None and len(pull_stream) > 0:
- kafkaProducer.sender(dsp_push_stream_result_topic, pull_stream["requestId"], pull_stream, 1)
- else:
- time.sleep(1)
- except Exception:
- logger.error("问题反馈异常：{}, requestId:{}", format_exc(), feedback.get("request_id"))
- logger.info("问题反馈线程执行完成")
|
