tuoheng_alg_move_20240715/concurrency/CommonProcess.py

# -*- coding: utf-8 -*-
import json
from multiprocessing import Process, Queue
from util import LogUtils
from util import KafkaUtils
from loguru import logger
from concurrency.FileUpdateThread import ImageFileUpdate
from concurrency.HeartbeatThread import Heartbeat
import time


class CommonProcess(Process):
    def __init__(self, fbQueue, hbQueue, content, msg, imageQueue, mode_service):
        super().__init__()
        self.fbQueue = fbQueue
        self.hbQueue = hbQueue
        self.content = content
        self.msg = msg
        self.mode_service = mode_service
        self.imageQueue = imageQueue


    def getFeedback(self):
        eBody = None
        try:
            eBody = self.fbQueue.get(block=False)
        except Exception as e:
            pass
        return eBody

        # 推送执行结果
    def sendImageResult(self, result):
        while self.imageQueue.full():
            logger.info("图片上传队列已满, 2秒后重试! requestId:{}", self.msg.get("request_id"))
            time.sleep(2)
        self.imageQueue.put(result)

    def run(self):
        # 初始化日志配置
        LogUtils.init_log(self.content)
        logger.info("心跳、图片上传，反馈进程开始执行, requestId:{}", self.msg.get("request_id"))
        # 启动心跳线程
        hb = Heartbeat(self.fbQueue, self.hbQueue, self.msg.get("request_id"), self.mode_service)
        hb.setDaemon(True)
        hb.start()
        # 图片上传线程
        imageFileUpdate = ImageFileUpdate(self.fbQueue, self.content, self.msg, self.imageQueue, self.mode_service)
        imageFileUpdate.setDaemon(True)
        imageFileUpdate.start()
        kafkaProducer = KafkaUtils.CustomerKafkaProducer(self.content, self.msg.get("request_id"))
        # 心跳线程检测
        heartbeat_num = 0
        # 图片上传线程检测
        imageFileUpdate_num = 0
        while True:
            try:
                if heartbeat_num == 0 and not hb.is_alive():
                    logger.error("未检测到心跳线程活动，心跳线程可能出现异常, reuqestId:{}", self.msg.get("request_id"))
                    break
                if imageFileUpdate_num == 0 and not imageFileUpdate.is_alive():
                    logger.error("未检测到图片上传线程活动，图片上传线程可能出现异常, reuqestId:{}", self.msg.get("request_id"))
                    break
                fb = self.getFeedback()
                if fb is not None and len(fb) > 0:
                    feedback = fb.get("feedback")
                    command = fb.get("command")
                    if feedback is not None and len(feedback) > 0:
                        kafkaProducer.get_producer()
                        kafkaProducer.sender(self.content["kafka"]["topic"]["dsp-alg-results-topic"],
                                             feedback["request_id"], feedback, 1)
                    if command is not None and len(command) > 0:
                        # 接收心跳线程和图片上传停止指令
                        if 'stop_heartbeat_imageFileUpdate' == command:
                            heartbeat_num += 1
                            imageFileUpdate_num += 1
                            hb.run_status = False
                            self.sendImageResult({"command": "stop"})
                            hb.join()
                            imageFileUpdate.join()
                            # 接收进程停止指令
                        if 'stop' == command:
                            heartbeat_num += 1
                            imageFileUpdate_num += 1
                            hb.run_status = False
                            self.sendImageResult({"command": "stop"})
                            hb.join()
                            imageFileUpdate.join()
                            break
                else:
                    time.sleep(1)
            except Exception as e:
                logger.exception("结果反馈异常：{}, requestId:{}", e, self.msg.get("request_id"))
        logger.info("心跳、图片上传，反馈进程执行完成, requestId:{}", self.msg.get("request_id"))