Python 2023 年 7 月 10 日

【Python】第一个多线程项目的简单实战教程

在Python中进行多线程编程是一种有效的方式来提高程序的性能和并发性。本项目旨在帮助您开始进行多线程编程，并提供一个简单的实战示例。在这个项目中，您将学习如何使用Python的threading模块创建和管理多个线程。通过将任务分配给不同的线程，可以同时执行多个任务，从而加快程序的执行速度。

Python 多线程

##多线程概述我们首先简单回顾一下多线程（我前面有文章讲过）多线程是指在一个程序中同时执行多个部分或多个任务，这些部分或任务可以并发地执行，提高程序的效率和性能。Python 提供了threading模块来实现多线程编程。

代码解析

在开篇部分，我们简要介绍了多线程的概念和优势。多线程是一种并发执行的方式，可以同时执行多个任务，提高程序的效率和性能。它通过充分利用计算机的多核处理器，在同一个程序中独立运行多个线程，从而实现任务的并发执行。

在我的项目里，我们将拆分几个代码片段来详解。这次我们用例是多线程的查询请求，然后多线程的 POST 给后端。

先看运行结果 微信截图_20230710103926.png

在代码里我们先创建一个数组，或者是请求能提取到数据的接口。

#数据示例
list = ['baicu.com','qq.com']

在代码里我选择了使用接口返回的数据处理成数组。

url = f"https://xxx.com?data={query_data}&page={current_page}"
response = requests.post(url)
domains = []
time.sleep(1.5)
try:
    if response.status_code == 200:
        json_data = response.json()
        if "site" in json_data and len(json_data["list"]) > 0:
            for item in json_data["list"]:
                domain = item["domain"]
                domains.append(domain)
            print("步骤1：", domains)
            if retries == max_retries:
                print(f"第 {current_page} 页获取失败")
except Exception as e:
    print("报错！", e)

在上述代码里，我我请求了一个接口返回的数据，我将每一组数据里面的domain提取出来，然后储存到我定义的 domains 数组里，使用 append方法。

###多线程执行因为我们每个数组里面有不固定的数量，所以写死线程数是不现实的。即使自定义，也会发生误差造成资源浪费。所以我们直接使用for循环来创建、关闭多线程。

threads = []  # 存储所有线程
for domain in domains:
    # 创建线程
    thread = threading.Thread(target=query_whois, args=(random.choice(whois_url), domain))
    # 启动线程
    thread.start()
    threads.append(thread)
# 等待线程完成
for thread in threads:
    thread.join()

这个时候我调用了query_whois方法。没错，这个就是我用来查询域名 whois 信息的方法，同时在方法里面调用 post 方法传递到后端。整个流程就完成了！


def query_whois(url, domain):
    response = requests.get(url.format(domain))
    data = json.loads(response.text)
    # print(data)
    if data["code"] == 200 and len(data) > 3:  # 使用相对等于判断
        whois_data = {
            "email": data.get("email", ""),
            "domain_reg_date": data.get("reg_date", ""),
            "LLC": data.get("LLC", ""),
        }
        set_domian(whois_data, domain)
    else:
        print("Whois 查询错误：code", data["code"], domain)

set_domian方法就是你写 post 到后端的。个人自定义即可

本文同步我的技术文档

作者：吃个甘蔗嚼一年发表日期：2023 年 7 月 10 日