kafka-python 执行两次初始化导致进程卡住-摩杜云开发者社区

Python logging库重复初始化导致进程卡住

前置知识

python的logging库
Python 的 logging 库是一个灵活且强大的日志记录工具，用于在应用程序中捕获、记录和处理日志信息。它提供了一种配置日志记录的方式，可以满足不同需求的应用程序。
以下是 logging 库的一些关键概念和组件：
Logger（记录器）：记录器是日志记录的入口点，负责发出各种日志消息。
Handler（处理器）：处理器将日志消息发送到目标，如控制台、文件或网络。
Formatter（格式化器）：格式化器定义日志输出的格式，用于美化和定制日志消息。
Level（级别）：级别用于控制日志消息的重要性，包括 DEBUG、INFO、WARNING、ERROR 和 CRITICAL。
Filter（过滤器）：过滤器允许更精细地控制哪些日志消息被记录。
配置文件：日志配置文件提供一种灵活的配置方式，允许通过文件而非代码进行日志配置。
python的celery框架
Celery 是一个开源的分布式任务队列系统，用于处理大量的异步任务。它允许你将任务从应用程序中分离出来，异步地执行它们，提高应用程序的性能和可伸缩性。Celery主要用于处理耗时的任务，如发送电子邮件、生成报告、处理图像等。
以下是 Celery 的一些主要特性和概念：
分布式任务队列： Celery 是一个分布式系统，用于处理异步任务，将任务分发到多个工作节点。
异步任务：允许将任务提交到队列，实现异步执行，提高应用性能和响应速度。
任务调度：支持定时任务调度，类似于 cron，可以在未来的特定时间执行任务。
消息代理：与多种消息代理（如 RabbitMQ、Redis、Amazon SQS）集成，用于在应用程序和工作节点之间传递任务消息。
结果存储：可将任务执行的结果保存在不同的后端存储中，例如数据库、缓存等。
任务重试：具备自动重试机制，可配置任务在失败时进行重试。
监控和管理：提供工具和界面用于监控和管理任务队列，包括 Web 界面和命令行工具。
多语言支持：主要用于 Python，但提供了多语言客户端库，支持其他编程语言的集成。
python连接kafka的库python-kakfa

kafka-python 是一个用于在 Python 中与 Apache Kafka 集成的客户端库。它提供了 KafkaProducer 类用于将消息发送到 Kafka 主题，以及 KafkaConsumer 类用于从 Kafka 主题中消费消息。通过这个库，你可以方便地在 Python 中与 Kafka 集群进行通信，实现消息的发布和订阅功能。kafka-python 还支持各种配置选项，允许你调整客户端的行为，以满足特定需求。

现象描述

python的celery启动后, celery worker 进程卡住, 无法处理任务
并且没有任何日志输出

原因概述

我们有一个代码仓库,

既有定时任务的代码, 又有Api应用的代码, 有同事加了一个定时任务, 不小心引入的Api的一个util代码, 导致执行了两次init_logger()初始化日志器方法,(一次定时任务本身的, 一次依赖链中的Api的)

init_logger()里面其中包含了一个初始化日志处理器(发送邮件)名为EmailHandler

EmailHandler里面初始化了kafkaProducer

但是执行第二次init_logger()的时候, 有这么一个逻辑, 执行了这个函数_clearExistingHandlers()

_clearExistingHandlers()这个函数的作用是清空已有的日志处理器列表

这个时间kafkaProducer刚刚在前一次初始化好相关资源, 还有相关的锁没有被释放

这个时候去清EmailHandler,就会导致那个锁没有释放, 无法创建第二个实例, 导致进程卡主没有日志

源码分析

/venv/lib/python3.7/site-packages/kafka/producer/kafka.py

line 445

def close(self, timeout=None):
    """Close the producer.

    This method sends a ProducerClose request to the Kafka broker and
    waits for acknowledgement. It is a good practice to call this method
    before the Python process exits to ensure any outstanding messages
    are delivered.

    Args:
        timeout (float): Maximum time to wait for the producer to complete
            the close operation, in seconds. If `None`, the method will
            block until the close is complete.

    Raises:
        KafkaException: If the producer close failed.
    """
    # ...

    # Acquire the lock to ensure thread safety during close
    with self._lock:
        # Check if the producer is already closed
        if self._closed:
            return

        # Set _closed flag to True to mark the producer as closed
        self._closed = True

    # ...

    # Wait for the background thread to finish
    if timeout is not None:
        self._sender_thread.join(timeout)

    # Release the lock
    with self._lock:
        # Check if the producer is already closed
        if self._closed:
            return

        # ...

    # ...

概括

with self._lock:：通过 with 语句，获取 _lock 锁，确保在多线程环境下的线程安全性。

if self._closed:：检查生产者是否已经关闭，如果已经关闭，直接返回，避免重复关闭。

self._closed = True：将 _closed 标志设置为 True，表示生产者已关闭。

self._sender_thread.join(timeout)：等待后台线程完成。_sender_thread 是一个在生产者初始化时启动的后台线程，负责异步发送消息到 Kafka broker。

with self._lock:：再次获取锁，确保在关闭期间不会有其他线程对生产者进行操作。

if self._closed:：再次检查生产者是否已经关闭，避免重复关闭。

此部分代码主要是为了确保在多线程环境下，对生产者的关闭操作是线程安全的，并等待后台线程完成。这有助于确保在关闭过程中不会出现竞态条件，从而确保生产者的关闭操作是可靠的。

排查步骤

由于我们的应用部署在华为云中, 所以日志使用的是华为云LTS, 而LTS没有采集到任何日志, 所以
手动进入k8s的pod中, 执行kubectl logs -f 查看日志, 还是什么日志也没有

然后执行了kubectl exec -it podname -n -- bash进入pod,

手动启动celery任务celery -A tasks.app worker -l 启动后打印了几行初始化日志后, 进程卡主, CTRL+C中断程序后, 打印出了错误的堆栈信息

kafka-python 执行两次初始化导致进程卡住_python

重现步骤

from kafka import producer
from config.config import ConfigInfo as Config
import time
bootstrap_servers = [f'{Config.KAFKA_QUOTE_MAIL_HOST1}:{Config.KAFKA_QUOTE_MAIL_PORT}',
                     f'{Config.KAFKA_QUOTE_MAIL_HOST2}:{Config.KAFKA_QUOTE_MAIL_PORT}',
                     f'{Config.KAFKA_QUOTE_MAIL_HOST3}:{Config.KAFKA_QUOTE_MAIL_PORT}']

kp = producer.KafkaProducer(bootstrap_servers=bootstrap_servers)

kp.close()

print('ok!')

这样就会报错, 如果close前面等待一段时间, 就不会报错

from kafka import producer
from config.config import ConfigInfo as Config
import time
bootstrap_servers = [f'{Config.KAFKA_QUOTE_MAIL_HOST1}:{Config.KAFKA_QUOTE_MAIL_PORT}',
                     f'{Config.KAFKA_QUOTE_MAIL_HOST2}:{Config.KAFKA_QUOTE_MAIL_PORT}',
                     f'{Config.KAFKA_QUOTE_MAIL_HOST3}:{Config.KAFKA_QUOTE_MAIL_PORT}']

kp = producer.KafkaProducer(bootstrap_servers=bootstrap_servers)
time.sleep(1)
kp.close()

print('ok!')

解决方案

避免重复执行kafkaPruducer的销毁和初始化

应用发版后, 不仅需要检查应用运行状态, 还要检查是否有日志输出