python多进程卡死排查记录

喜欢ヅ旅行 2022-09-04 11:50 468阅读 0赞

**问题背景：**  
        python程序一开始为单进程结构，属于CPU密集型计算任务，为提高性能调整为多进程并行计算。  
        调整后运行时程序启动一会就全部进程卡死，不再往下运行。

**问题排查：**  
        程序中没有使用到进程锁或者分布式锁，所以先是排除了死锁的原因，可能为逻辑BUG或者代码异常。  
        多进程下只好通过打日志的方式定位下程序运行到什么位置了，最后定位出不再往下运行的代码行为：

msgs: Dict = kwargs['msgs']

这部分的完整代码块为：

class ConclusionWriteHandler(SinkWriterInterface):
        """数据分析结论入MQ handler"""
    
        def write(self, **kwargs):
            msgs: Dict = kwargs['msgs']
            serializer = kwargs['serializer']
            kafka_client = Kafka()
            for conclusion_category, conclusion_list in msgs.items():
                if not conclusion_list:
                    continue
                sub_msgs = [serializer(x) for x in conclusion_list]
                # 分析结论的topic名称和结论类型是保持一致的
                kafka_client.produce(topic=conclusion_category, msgs=sub_msgs)
                log.info(f'load {len(msgs)} conclusions to topic {conclusion_category}')

一开始看到定位出的位置是一脸懵逼的，一个普通的字典取值操作怎么就卡住了，以为是数据格式存在问题，添加 try catch 尝试捕获异常，但并没有触发。  
        当前是在pycharm内使用debug模式运行的程序，确认这块代码没有问题后，尝试在命令行运行，看下定位出的位置是不是一致的，果然，命令行下运行这行代码(msgs: Dict = kwargs\['msgs'\])运行通过，于是继续添加日志定位程序卡住的位置，随后定位到了kafka数据发送模块:

def produce(self, topic: str, msgs: List) -> NoReturn:
        """
        生产者
    
        Args:
            topic: topic
            msgs: msg列表
    
        """
        for msg in msgs:
            self.producer.poll(0)
            self.producer.produce(topic=topic, value=msg.encode('utf-8'), callback=self.delivery_report)
        self.producer.flush()

发现数据都有发送，但是在运行到 self.producer.flush() 时程序不再运行了，推测为 confluent-kafka 这个位置的使用出现了问题，其中kafka客户端代码为：

class KafkaCli(metaclass=SingletonClass):
        def __init__(self):
            self.__producer = None
            self.__consumer = None
    
        def client_init(self, bootstrap_servers: str,
                        group_id: str,
                        session_timeout_ms: int = 6000,
                        enable_auto_commit: bool = True,
                        auto_offset_reset: str = 'earliest') -> NoReturn:
            """producer && consumer init"""
            producer_config = {
                'bootstrap.servers': bootstrap_servers
            }
            consumer_config = dict(**producer_config,  **{
                'group.id': group_id,
                'session.timeout.ms': session_timeout_ms,
                'enable.auto.commit': enable_auto_commit,
                # 'on_commit': on_commit,
                'auto.offset.reset': auto_offset_reset
            })
            self.__consumer = kafka.Consumer(consumer_config)
            self.__producer = kafka.Producer(producer_config)

这里kafka client为单例模式，并且的程序初始化的时间初始化 producer 和 consumer，其中单例元类为:

class SingletonClass(type):
        """单例实现-元类"""
        _instance_lock = threading.Lock()
    
        def __call__(cls, *args, **kwargs):
            if not hasattr(cls, '_instance'):
                with cls._instance_lock:
                    if not hasattr(cls, '__instance'):
                        cls._instance = super().__call__(*args, **kwargs)
            return cls._instance

理论上，创建多进程的时间，主进程的所有资源会拷贝一份到子进程中，虽然这里在初始化时为单例对象，但创建的各个子进程仍然有自己的一个kafka client对象(这个也通过打印获取的kafka client地址确定了)        ![20210813160511114.png][]

所以回过头来看下程序卡住的位置：       ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70][]

**         producer.flush() 的作用是将生成者队列中的所有消息发送出去，这里卡住可能就是发送机制上出现问题了。**

**        面向搜索引擎解决问题：       **![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 1][]

链接地址: [https://github.com/confluentinc/confluent-kafka-python/issues/351][https_github.com_confluentinc_confluent-kafka-python_issues_351]

问题原因也看到了: kafka客户端使用了内部线程，创建子进程时并不会把主进程中的子线程也拷贝的子进程空间内。

![20210813160806644.png][]

**这其实和在POSIX(以前称为 Unix 的标准)上创建子进程的方式有关:**

**        1. 使用fork()系统调用创建进程的副本。**

**        2. 子进程使用execve()系统调用（或其变体之一，例如execl()）将自身替换为不同的程序。**

**我们都知道创建子进程会拷贝主进程中的所有内存内容，但实际上fork()并没有拷贝所有内容，其中就包括父进程中的线程，即：**

**        任何在父进程中运行的线程都不存在于子进程中**

所以程序卡住的原因也就找到了:

** kafka client是在程序启动的时间做的初始化，底层使用的是线程，所以接下来创建出的子进程中不含有kafka client底层需要的线程，最后producer在发送数据是没有底层线程进行调度，直接阻塞了。**

**问题修复：**

**将kafka client初始化的工作在各个子进程中完成。**

class KafkaCli(metaclass=KafkaSingletonClass):
        def __init__(self):
            self.__producer = None
            self.__consumer = None
            self.client_init(
                bootstrap_servers=CONFIG.KAFKA.bootstrap_servers,
                group_id=CONFIG.KAFKA.group_id,
                session_timeout_ms=CONFIG.KAFKA.session_timeout_ms,
                enable_auto_commit=CONFIG.KAFKA.enable_auto_commit,
                auto_offset_reset=CONFIG.KAFKA.auto_offset_reset
            )
    
        def client_init(self, bootstrap_servers: str,
                        group_id: str,
                        session_timeout_ms: int = 6000,
                        enable_auto_commit: bool = True,
                        auto_offset_reset: str = 'earliest') -> NoReturn:
            """producer && consumer init"""
            producer_config = {
                'bootstrap.servers': bootstrap_servers
            }
            consumer_config = dict(**producer_config,  **{
                'group.id': group_id,
                'session.timeout.ms': session_timeout_ms,
                'enable.auto.commit': enable_auto_commit,
                # 'on_commit': on_commit,
                'auto.offset.reset': auto_offset_reset
            })
            self.__consumer = kafka.Consumer(consumer_config)
            self.__producer = kafka.Producer(producer_config)

同时修改使用的单例元类，实现每个进程内的单例：

class KafkaSingletonClass(type):
        """kafka 专用的单例元类"""
        _instance_lock = threading.Lock()
        _instance = {}
    
        def __call__(cls, *args, **kwargs):
            pid = mp.current_process().pid
            if pid not in cls._instance:
                with cls._instance_lock:
                    if pid not in cls._instance:
                        cls._instance[pid] = super().__call__(*args, **kwargs)
            return cls._instance[pid]

问题修复只改动了kafka client初始化部分的代码，不需要对其它位置进行调整。

**子进程创建中的线程拷贝问题相关链接:**

[        https://www.linuxprogrammingblog.com/threads-and-fork-think-twice-before-using-them][https_www.linuxprogrammingblog.com_threads-and-fork-think-twice-before-using-them]

[        https://pythonspeed.com/articles/python-multiprocessing/][https_pythonspeed.com_articles_python-multiprocessing]

[20210813160511114.png]: /images/20220829/04b828dcd0484146b2d948b17e0aa741.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70]: /images/20220829/b094a8dfd07d4d6b90a54da12f00d897.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 1]: /images/20220829/dd63507cee204e3b98a798e9384fcca3.png
[https_github.com_confluentinc_confluent-kafka-python_issues_351]: https://github.com/confluentinc/confluent-kafka-python/issues/351
[20210813160806644.png]: /images/20220829/e124d523fc3d44f580fb22ac8889b361.png
[https_www.linuxprogrammingblog.com_threads-and-fork-think-twice-before-using-them]: https://www.linuxprogrammingblog.com/threads-and-fork-think-twice-before-using-them
[https_pythonspeed.com_articles_python-multiprocessing]: https://pythonspeed.com/articles/python-multiprocessing/