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

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

### 前置知识

1. python的logging库

Python 的 logging 库是一个灵活且强大的日志记录工具，用于在应用程序中捕获、记录和处理日志信息。它提供了一种配置日志记录的方式，可以满足不同需求的应用程序。

以下是 logging 库的一些关键概念和组件：

Logger（记录器）： 记录器是日志记录的入口点，负责发出各种日志消息。

Handler（处理器）： 处理器将日志消息发送到目标，如控制台、文件或网络。

Formatter（格式化器）： 格式化器定义日志输出的格式，用于美化和定制日志消息。

Level（级别）： 级别用于控制日志消息的重要性，包括 DEBUG、INFO、WARNING、ERROR 和 CRITICAL。

Filter（过滤器）： 过滤器允许更精细地控制哪些日志消息被记录。

配置文件： 日志配置文件提供一种灵活的配置方式，允许通过文件而非代码进行日志配置。

2. python的celery框架

Celery 是一个开源的分布式任务队列系统，用于处理大量的异步任务。它允许你将任务从应用程序中分离出来，异步地执行它们，提高应用程序的性能和可伸缩性。Celery主要用于处理耗时的任务，如发送电子邮件、生成报告、处理图像等。

以下是 Celery 的一些主要特性和概念：

分布式任务队列： Celery 是一个分布式系统，用于处理异步任务，将任务分发到多个工作节点。

异步任务： 允许将任务提交到队列，实现异步执行，提高应用性能和响应速度。

任务调度： 支持定时任务调度，类似于 cron，可以在未来的特定时间执行任务。

消息代理： 与多种消息代理（如 RabbitMQ、Redis、Amazon SQS）集成，用于在应用程序和工作节点之间传递任务消息。

结果存储： 可将任务执行的结果保存在不同的后端存储中，例如数据库、缓存等。

任务重试： 具备自动重试机制，可配置任务在失败时进行重试。

监控和管理： 提供工具和界面用于监控和管理任务队列，包括 Web 界面和命令行工具。

多语言支持： 主要用于 Python，但提供了多语言客户端库，支持其他编程语言的集成。

3. 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 <br>

line 445

```python

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

# ...

# ...

```

概括

```python

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, <br>

手动启动celery任务`celery -A tasks.app worker -l`

启动后打印了几行初始化日志后, 进程卡主, CTRL+C中断程序后, 打印出了错误的堆栈信息

### 重现步骤

```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前面等待一段时间, 就不会报错

```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)

time.sleep(1)

kp.close()

print('ok!')

```

### 解决方案

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

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