Python多线程并发实例及其优化

Python的多线程并发实例，在处理IO密集型任务时，可以有效提升程序的执行效率。在本文中，我们将通过两个示例来详细讲解Python的多线程并发实现及其优化方法。

示例一

需求

编写一个程序，使用多线程并发实现下载多个图片，并通过回调函数显示已下载的图片数量。

实现过程

1. 安装依赖库

使用Python的requests库来实现图片的下载操作，使用tqdm库来实现进度条的显示。

pip install requests
pip install tqdm

2. 构建单个图片下载函数

定义一个下载单个图片的函数downloadImage，该函数接受图片的URL地址作为参数，并返回图片的二进制数据。如果URL地址无法访问，则抛出异常。

import requests

def downloadImage(url):
    response = requests.get(url)
    if response.status_code != 200:
        raise Exception("Image download failed")

    return response.content

3. 构建多线程下载函数

定义一个下载多个图片的函数downloadImages，该函数接受一个表示图片URL地址列表的参数，并使用多线程并发的方式，下载所有的图片并返回已下载图片的个数。

import threading
from tqdm import tqdm

def downloadImages(urls, callback):
    count = 0

    def downloadTask(url):
        nonlocal count
        try:
            downloadImage(url)
            count += 1
            callback(count)
        except Exception as e:
            print(f"Error while downloading image from {url}: {e}")

    tasks = []
    for url in urls:
        task = threading.Thread(target=downloadTask, args=(url,))
        tasks.append(task)
        task.start()

    for task in tasks:
        task.join()

    return count

在downloadImages函数中，首先定义一个表示已下载图片个数的计数器count，以及一个downloadTask函数，downloadTask函数用于下载单个图片，并在下载完成后将计数器+1并调用回调函数callback来显示已下载图片的数量。

然后，使用线程池的方式创建多个downloadTask任务，并启动它们。在所有任务完成后，返回已完成任务的数量。

4. 测试

在程序中调用downloadImages函数，传入图片URL地址列表以及回调函数，来下载所有图片并显示已下载图片数量。

urls = [
    'https://www.python.org/static/apple-touch-icon-144x144-precomposed.png',
    'https://www.python.org/static/apple-touch-icon-72x72-precomposed.png',
    'https://www.python.org/static/apple-touch-icon-57x57-precomposed.png',
    'https://www.python.org/static/community_logos/python-powered-w-200x80.png',
]

def callback(count):
    print(f"{count} images downloaded")

count = downloadImages(urls, callback)
print(f"Total {count} images downloaded")

优化过程

1. 优化下载函数

使用requests库下载图片时会自动进行连接池管理和HTTP协议版本选择等优化，因此不需要再对下载函数进行优化。

2. 优化线程数目

使用过多的线程会导致线程之间的切换频繁，从而影响程序执行效率。因此，我们需要根据实际情况来选择合适的线程数目。

在示例一中，可以通过修改线程数目来测试不同线程数目下的性能表现。

示例二

需求

编写一个程序，使用多线程并发实现计算大量数组的平均数，并通过回调函数显示已计算完成的数组个数。

实现过程

1. 构建计算平均数的函数

定义一个计算数组平均数的函数calcAverage，该函数接受一个数组作为参数，并返回该数组的平均数。

import random

def calcAverage(nums):
    return sum(nums) / len(nums)

为了模拟大量的数组计算，这里可以使用Python的random模块，生成包含1000个随机数的数组。

data = [random.random() for _ in range(1000)]

2. 构建多线程计算函数

定义一个多线程并发计算函数calcAverages，该函数接受一个表示数组列表的参数，并使用多线程并发的方式，计算每个数组的平均数，并返回已计算完成数组的个数。

import threading

def calcAverages(arrays, callback):
    count = 0

    def calcTask(nums):
        nonlocal count
        calcAverage(nums)
        count += 1
        callback(count)

    tasks = []
    for nums in arrays:
        task = threading.Thread(target=calcTask, args=(nums,))
        tasks.append(task)
        task.start()

    for task in tasks:
        task.join()

    return count

在calcAverages函数中，首先定义一个表示已完成计算的数组个数的计数器count，以及一个calcTask函数，calTask函数用于计算单个数组的平均数，并在计算完成后将计数器+1并调用回调函数callback来显示已计算数组的数量。

然后，使用线程池的方式创建多个calcTask任务，并启动它们。在所有任务完成后，返回已完成任务的数量。

3. 测试

在程序中调用calcAverages函数，传入包含多个数组的数组列表以及回调函数，来计算所有数组的平均数并显示已计算完成数组数量。

arrays = [[random.random() for _ in range(100)] for _ in range(100)]
def callback(count):
    print(f"{count} arrays calculated")

count = calcAverages(arrays, callback)
print(f"Total {count} arrays calculated")

优化过程

1. 优化计算函数

在示例二中，如果要计算的数组数目非常大，那么计算平均数这种密集计算的任务可能会变成程序的瓶颈。因此，可以考虑使用numpy库来进行计算，以达到更好的性能。

import numpy as np

def calcAverage(nums):
    return np.mean(nums)

2. 优化线程数目

同样可以通过修改线程数目来测试不同线程数目下的性能表现。但是，和示例一不同的是，在示例二中由于多个线程都需要读取数据，同时进行CPU密集型的计算任务，因此线程数量不能过多，否则会出现线程间竞争且CPU占用率较高的问题。