# Python多进程编程详解：一个实用的任务分发与处理示例

在现代计算中，利用多核CPU的能力来提高程序的性能已成为必不可少的技能。Python的`multiprocessing`模块提供了简便的接口来实现多进程并行计算。本文将通过一个实际的代码示例，详细介绍如何使用Python的多进程来分发和处理任务。



## 引言

在处理大量任务时，单线程或单进程往往无法充分利用多核CPU的性能。通过多进程并行处理，可以显著提高程序的执行效率。下面，我们将介绍一个完整的代码示例，演示如何使用`multiprocessing`模块实现任务的并行处理。

## 代码结构概览

首先，让我们先看一下代码的整体结构：

```python
import logging
import time
from multiprocessing import cpu_count, get_context

def process_task(task_queue, enable_trigger):
    # 处理任务的工作进程函数
    pass

def distribute_tasks(task_queue, batch_size, incremmental, task_size):
    # 分发任务到任务队列
    pass

def main():
    # 主函数，设置多进程环境并启动任务处理
    pass

if __name__ == '__main__':
    main()
```

代码主要包含以下几个部分：

1. **导入必要的模块**：包括`logging`、`time`和`multiprocessing`。
2. **定义工作进程函数**：`process_task`，用于处理从任务队列中获取的任务。
3. **定义任务分发函数**：`distribute_tasks`，用于将任务放入任务队列。
4. **主函数**：`main`，设置多进程环境，启动工作进程，并分发任务。

## 详细代码解析

接下来，我们将逐步解析代码的每个部分。

### 1. 导入模块

```python
import logging
import time
from multiprocessing import cpu_count, get_context
```

- `logging`：用于记录日志信息，方便调试和监控程序运行。
- `time`：用于记录时间，计算程序运行耗时。
- `multiprocessing`：核心模块，提供创建多进程所需的功能。

### 2. 定义工作进程函数 `process_task`

```python
def process_task(task_queue, enable_trigger):
    while True:
        task = task_queue.get()
        if task is None:
            # 接收到退出信号，结束循环
            task_queue.task_done()
            break
        # 处理任务（此处为占位符，实际逻辑需自行实现）
        logging.info(f"Processing task: {task}")
        time.sleep(0.1)  # 模拟处理时间
        task_queue.task_done()
```

**功能说明**：

- 不断从`task_queue`中获取任务。
- 如果接收到`None`，表示需要结束进程，跳出循环。
- 否则，处理任务（此处用`time.sleep(0.1)`模拟处理时间）。
- 调用`task_queue.task_done()`，表示任务已处理完毕。

**关键点**：

- 使用`while True`循环，使进程持续运行，直到接收到退出信号。
- 使用`task_queue.get()`从队列中获取任务，队列是进程间通信的主要方式。
- 通过`task_queue.task_done()`通知队列，任务已完成。

### 3. 定义任务分发函数 `distribute_tasks`

```python
def distribute_tasks(task_queue, batch_size, incremmental, task_size):
    # 将任务分发到任务队列中
    for i in range(batch_size):
        task_queue.put(f"Task {i}")
    logging.info(f"Distributed {batch_size} tasks.")
```

**功能说明**：

- 将一批任务放入`task_queue`中，供工作进程消费。
- `batch_size`决定了任务的数量。

**关键点**：

- 使用`task_queue.put()`将任务放入队列。
- 任务可以是任何可序列化的对象，这里使用字符串表示。

### 4. 主函数 `main`

```python
def main():
    start = time.time()
    max_workers = 0
    enable_trigger = False
    incremmental = True  # 假设这是一个布尔标志

    num_processor = int(max_workers) or cpu_count()
    task_size = num_processor * 3
    ctx = get_context('spawn')  # 使用'spawn'以兼容Windows
    batch_size = 1000

    logging.basicConfig(level=logging.INFO)
    logging.info(
        f"[start update candidate resume lang]: "
        f"use processors {num_processor}, "
        f"task_length {task_size}"
    )

    task_queue = ctx.JoinableQueue(maxsize=task_size)
    # 创建工作进程
    workers = []
    for _ in range(num_processor):
        p = ctx.Process(target=process_task, args=(task_queue, enable_trigger))
        p.start()
        workers.append(p)

    # 分发任务
    distribute_tasks(task_queue, batch_size, incremmental, task_size)

    # 等待所有任务完成
    task_queue.join()

    # 发送信号以停止工作进程
    for _ in workers:
        task_queue.put(None)

    # 等待所有工作进程结束
    for p in workers:
        p.join()

    logging.info(f"Finished processing in {time.time() - start} seconds.")
```

**功能说明**：

- 设置多进程环境，创建工作进程。
- 分发任务到任务队列。
- 等待任务完成，发送结束信号，关闭进程。

**关键点详解**：

#### a. 初始化参数

```python
start = time.time()
max_workers = 0
enable_trigger = False
incremmental = True  # 假设这是一个布尔标志
```

- `start`：记录开始时间，用于计算总耗时。
- `max_workers`：最大工作进程数，默认为0。
- `enable_trigger`和`incremmental`：控制流程的标志变量。

#### b. 确定进程数量和任务队列大小

```python
num_processor = int(max_workers) or cpu_count()
task_size = num_processor * 3
ctx = get_context('spawn')  # 使用'spawn'以兼容Windows
batch_size = 1000
```

- `num_processor`：如果`max_workers`为0，则使用`cpu_count()`获取CPU核心数。
- `task_size`：任务队列的最大大小，设置为进程数的3倍。
- `ctx`：获取上下文，这里使用`spawn`以兼容Windows操作系统。
- `batch_size`：一次分发的任务数量。

#### c. 配置日志

```python
logging.basicConfig(level=logging.INFO)
logging.info(
    f"[start update candidate resume lang]: "
    f"use processors {num_processor}, "
    f"task_length {task_size}"
)
```

- 配置日志级别为`INFO`，用于输出信息。
- 记录使用的处理器数量和任务长度。

#### d. 创建任务队列和工作进程

```python
task_queue = ctx.JoinableQueue(maxsize=task_size)
# 创建工作进程
workers = []
for _ in range(num_processor):
    p = ctx.Process(target=process_task, args=(task_queue, enable_trigger))
    p.start()
    workers.append(p)
```

- 创建一个可加入的任务队列`JoinableQueue`，指定最大大小`maxsize`。
- 使用`Process`创建工作进程，目标函数为`process_task`，并传入参数。
- 启动进程并将其添加到`workers`列表中。

#### e. 分发任务

```python
distribute_tasks(task_queue, batch_size, incremmental, task_size)
```

- 调用`distribute_tasks`函数，将任务放入任务队列中。

#### f. 等待任务完成并关闭进程

```python
# 等待所有任务完成
task_queue.join()

# 发送信号以停止工作进程
for _ in workers:
    task_queue.put(None)

# 等待所有工作进程结束
for p in workers:
    p.join()
```

- `task_queue.join()`：阻塞主进程，直到队列中的所有任务都被处理完毕。
- 通过向每个工作进程发送`None`，通知它们退出循环，结束进程。
- 使用`p.join()`等待所有进程结束，确保资源被正确释放。

#### g. 输出总耗时

```python
logging.info(f"Finished processing in {time.time() - start} seconds.")
```

- 计算并输出程序的总运行时间。

### 5. 运行主函数

```python
if __name__ == '__main__':
    main()
```

- 确保只有在直接运行脚本时才执行`main()`函数。
- 这是Python的惯用法，防止在被导入时意外执行代码。

## 总结

通过上述代码，我们实现了一个简单而实用的多进程任务处理程序。关键点包括：

- 使用`multiprocessing`模块的`Process`和`JoinableQueue`来实现进程间通信和任务分发。
- 使用`get_context('spawn')`来兼容不同的操作系统，特别是Windows。
- 通过发送特殊的退出信号（如`None`），优雅地结束工作进程。
- 使用`task_queue.join()`和`task_queue.task_done()`来同步任务的完成情况。

## 实际应用建议

- **任务设计**：在实际应用中，任务可能是需要大量计算或IO操作的函数。应将实际的处理逻辑替换占位符部分。
- **异常处理**：在生产环境中，需添加异常处理机制，确保进程异常退出时不会导致死锁或资源泄漏。
- **资源管理**：注意进程和队列等资源的正确关闭和释放，防止内存泄漏。
- **性能调优**：根据任务的复杂度和系统资源，适当调整`num_processor`和`task_size`等参数。


## 整体程序


```python
import logging
import time
from multiprocessing import cpu_count, get_context

def process_task(task_queue, enable_trigger):
    while True:
        task = task_queue.get()
        if task is None:
            # Signal to exit
            task_queue.task_done()
            break
        # Process the task (placeholder for actual processing logic)
        logging.info(f"Processing task: {task}")
        time.sleep(0.1)  # Simulate processing time
        task_queue.task_done()

def distribute_tasks(task_queue, batch_size, incremmental, task_size):
    # Distribute tasks into the task_queue
    for i in range(batch_size):
        task_queue.put(f"Task {i}")
    logging.info(f"Distributed {batch_size} tasks.")

def main():
    start = time.time()
    max_workers = 0
    enable_trigger = False
    incremmental = True  # Assuming this is a boolean flag

    num_processor = int(max_workers) or cpu_count()
    task_size = num_processor * 3
    ctx = get_context('spawn')  # Use 'spawn' for Windows compatibility
    batch_size = 1000

    logging.basicConfig(level=logging.INFO)
    logging.info(
        f"[start update candidate resume lang]: "
        f"use processors {num_processor}, "
        f"task_length {task_size}"
    )

    task_queue = ctx.JoinableQueue(maxsize=task_size)
    # Create worker processes
    workers = []
    for _ in range(num_processor):
        p = ctx.Process(target=process_task, args=(task_queue, enable_trigger))
        p.start()
        workers.append(p)

    # Corrected function call with missing comma
    distribute_tasks(task_queue, batch_size, incremmental, task_size)

    # Wait for all tasks to be processed
    task_queue.join()

    # Send a signal to stop the workers
    for _ in workers:
        task_queue.put(None)

    # Properly close the pool
    for p in workers:
        p.join()

    logging.info(f"Finished processing in {time.time() - start} seconds.")

if __name__ == '__main__':
    main()

```

Python多进程编程详解：一个实用的任务分发与处理示例

```python
import json

# 读取 1.json 文件中的数据
with open('1.json', 'r', encoding='utf-8') as file1:
    dict1 = json.load(file1)

# 读取 2.json 文件中的数据
with open('2.json', 'r', encoding='utf-8') as file2:
    dict2 = json.load(file2)

# 合并字典
merged_dict = {**dict1, **dict2}  # 使用解包运算符合并字典

# 将合并后的字典写入 3.json 文件
with open('3.json', 'w', encoding='utf-8') as file3:
    json.dump(merged_dict, file3, ensure_ascii=False, indent=4)

print("字典合并成功，保存为 3.json")

```

Python如何优雅合并字典

Docker lmdeploy 快速部署Qwen2.5模型openai接口

直接从 Docker Hub 拉取：

```bash
docker pull openmmlab/lmdeploy:latest
```




等效于：

```bash
docker pull docker.io/openmmlab/lmdeploy:latest
```

一些可以用的镜像加速：https://gist.github.com/y0ngb1n/7e8f16af3242c7815e7ca2f0833d3ea6#


使用方法1，编辑或创建 /etc/docker/daemon.json 文件，添加以下内容：


```bash
{
  "registry-mirrors": ["https://dockerproxy.com"]
}
```
保存并重启 Docker 服务：

```bash
sudo systemctl daemon-reload
sudo systemctl restart docker
```

使用方法2，更加推荐用方法2 ，直接加前缀：


```bash
docker pull dockerproxy.com/openmmlab/lmdeploy:latest
```

别的镜像也可以用，比如dockerpull.org，用法一样：
```bash
docker pull dockerpull.org/openmmlab/lmdeploy:latest
```

Docker镜像加速办法，docker pull不求人

```python
import json
import random
from concurrent.futures import ProcessPoolExecutor, as_completed
from tqdm import tqdm
import requests

def get_api_response(query, headers, url, model, prompt_guanfang, style_zhongwen):
    """
    模拟API请求，发送一个POST请求并返回响应。
    """
    payload = {
        "query": query,
        "model": model,
        "prompt_guanfang": prompt_guanfang,
        "style_zhongwen": style_zhongwen
    }
    try:
        response = requests.post(url, json=payload, headers=headers)
        return response.json()  # 假设API返回JSON数据
    except requests.exceptions.RequestException as e:
        return {"error": str(e)}

def worker(args):
    """
    Worker函数，用于多进程执行。
    """
    query, url, headers, model, prompt_guanfang, style_zhongwen = args
    return get_api_response(query, headers, url, model, prompt_guanfang, style_zhongwen)

if __name__ == '__main__':
    # 定义所有可用的API端口
    ports = list(range(8001, 8009))  # 8001到8008
    urls = [f"http://127.0.0.1:{port}/v1/chat/completions" for port in ports]

    headers = {
        "Content-Type": "application/json",
    }

    model = "gpt-4o-mini"  # 可以根据需要调整

    # 读取 JSON 文件内容（这里假设风格和查询数据以JSON文件存储）
    with open('风格翻译.json', 'r', encoding='utf-8') as json_file:
        style_dict = json.load(json_file)

    with open('用户原始输入.json', 'r', encoding='utf-8') as json_file:
        user_in = json.load(json_file)

    # 假设load_queries函数返回所有查询的列表
    def load_queries(user_in):
        return [entry["query"] for entry in user_in]

    queries = load_queries(user_in)
    queries = random.sample(queries, 200)  # 随机选择200条查询进行处理

    # 初始化结果字典
    responses = []

    # 准备多进程池
    num_processes = len(urls)  # 进程数等于可用URL数量
    with ProcessPoolExecutor(max_workers=num_processes) as executor:
        # 准备任务列表，将每个查询分配到一个URL（轮询分配）
        tasks = []
        idx_url = 0
        for idx, query in enumerate(queries):
            for key in style_dict:
                prompt_guanfang = style_dict[key]["prompt_guanfang"]
                style_zhongwen = style_dict[key]["style_zhongwen"]

                url = urls[idx_url % len(urls)]  # 轮询分配URL
                idx_url += 1
                tasks.append((query, url, headers, model, prompt_guanfang, style_zhongwen))

        # 使用tqdm显示进度条
        futures = {executor.submit(worker, task): task for task in tasks}
        for future in tqdm(as_completed(futures), total=len(futures), desc="处理查询"):
            response = future.result()
            responses.append(response)

    # 保存所有响应到JSON文件
    with open('result.json', 'w', encoding='utf-8') as json_file:
        json.dump(responses, json_file, ensure_ascii=False, indent=4)

    print("所有API请求处理完毕，结果已保存至result.json")

```

利用Python的多进程加速API请求

我是一个专注技术分享的博主,尤其爱搞深度学习和单片机电子。你现在看到的是我的个人博客网站。