python3多线程实战（python3经典编程案例）

骑猪看日落 2022-09-08 05:59 230阅读 0赞

1.  [python3多进程实战（python3经典编程案例）][python3_python3]
2.  [python3多线程实战（python3经典编程案例）][python3_python3 1]
3.  [python3 协程实战（python3经典编程案例）][python3 _python3]

总结：

*  python多线程适用在I/O密集型的任务中。对于I/O密集型任务来说，较少的时间用在cpu计算上，较多的时间用在I/O上，如文件读写，web请求，数据库请求等；
 *  对于计算密集型任务，应该使用多进程。

## 一. 多线程任务对比 ##

线程也是轻量级进程，是操作系统能够进行运算调度的最小单位。它被包含在进程中，是进程的实际运作单位

一个线程可以创建和撤销另一个线程，同一进程的多个线程之间可以并发执行。

线程有就绪，阻塞，运行3中基本状态。

计算密集型任务-多进程

from multiprocessing import Process
    import os, time
    
    #计算密集型任务
    def work():
        res = 0
        for i in range(100000000):
            res *= i
    
    if __name__ == "__main__":
        l = []
        print("本机为",os.cpu_count(),"核 CPU")  # 本机为4核
        start = time.time()
        for i in range(4):
            p = Process(target=work)  # 多进程
            l.append(p)
            p.start()
        for p in l:
            p.join()
        stop = time.time()
        print("计算密集型任务，多进程耗时 %s" % (stop - start))

计算密集型任务-多线程

from threading import Thread
    import os, time
    
    #计算密集型任务
    def work():
        res = 0
        for i in range(100000000):
            res *= i
    
    if __name__ == "__main__":
        l = []
        print("本机为",os.cpu_count(),"核 CPU")  # 本机为4核
        start = time.time()
        for i in range(4):
            p = Thread(target=work)  # 多进程
            l.append(p)
            p.start()
        for p in l:
            p.join()
        stop = time.time()
        print("计算密集型任务，多线程耗时 %s" % (stop - start))

I/O密集型任务-多进程

from multiprocessing import Process
    import os, time
    
    #I/0密集型任务
    def work():
        time.sleep(2)
        print("===>", file=open("tmp.txt", "w"))
    
    if __name__ == "__main__":
        l = []
        print("本机为", os.cpu_count(), "核 CPU")  # 本机为4核
        start = time.time()
        for i in range(400):
            p = Process(target=work)  # 多进程
            l.append(p)
            p.start()
        for p in l:
            p.join()
        stop = time.time()
        print("I/0密集型任务，多进程耗时 %s" % (stop - start))

I/O密集型任务-多线程

from threading import Thread
    import os, time
    
    #I/0密集型任务
    def work():
        time.sleep(2)
        print("===>", file=open("tmp.txt", "w"))
    
    
    if __name__ == "__main__":
        l = []
        print("本机为", os.cpu_count(), "核 CPU")  # 本机为4核
        start = time.time()
    
        for i in range(400):
            p = Thread(target=work)  # 多线程
            l.append(p)
            p.start()
        for p in l:
            p.join()
        stop = time.time()
        print("I/0密集型任务，多线程耗时 %s" % (stop - start))

结论：在python中，对于计算密集型任务，多进程占优势；对于IO密集型任务，多线程占优势。

## 二. threading模块 ##

### 2.1 通过实例化threading.Thread类来创建线程 ###

调用start()方法。

import time
    import threading
    
    
    def task_thread(counter):
        print(
            f'线程名称：{ threading.current_thread().name} 参数：{ counter} 开始时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}'
        )
        num = counter
        while num:
            time.sleep(3)
            num -= 1
        print(
            f'线程名称：{ threading.current_thread().name} 参数：{ counter} 结束时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}'
        )
    
    
    if __name__ == "__main__":
        print(f'主线程开始时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')
    
        # 初始化3个线程，传递不同的参数
        t1 = threading.Thread(target=task_thread, args=(3,))
        t2 = threading.Thread(target=task_thread, args=(2,))
        t3 = threading.Thread(target=task_thread, args=(1,))
        # 开启三个线程
        t1.start()
        t2.start()
        t3.start()
        # 等待运行结束
        t1.join()
        t2.join()
        t3.join()
    
        print(f'主线程结束时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')

### 2.2 继承Thread类创建线程 ###

在子类中重写`run()`和`init()`方法

import time
    import threading
    
    
    class MyThread(threading.Thread):
        def __init__(self, counter):
            super().__init__()
            self.counter = counter
    
    
        def run(self):
    
            print(
                f'线程名称：{ threading.current_thread().name} 参数：{ self.counter} 开始时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}'
            )
            counter = self.counter
            while counter:
                time.sleep(3)
                counter -= 1
            print(
                f'线程名称：{ threading.current_thread().name} 参数：{ self.counter} 结束时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}'
            )
    
    
    if __name__ == "__main__":
        print(f'主线程开始时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')
    
        # 初始化3个线程，传递不同的参数
        t1 = MyThread(3)
        t2 = MyThread(2)
        t3 = MyThread(1)
        # 开启三个线程
        t1.start()
        t2.start()
        t3.start()
        # 等待运行结束
        t1.join()
        t2.join()
        t3.join()
    
        print(f'主线程结束时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')

### 2.3 继承Thread类调用外部传入参数 ###

import time
    import threading
    
    
    def task_thread(counter):
        print(f'线程名称：{ threading.current_thread().name} 参数：{ counter} 开始时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')
        num = counter
        while num:
            time.sleep(3)
            num -= 1
        print(f'线程名称：{ threading.current_thread().name} 参数：{ counter} 结束时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')
    
    
    class MyThread(threading.Thread):
        def __init__(self, target, args):
            super().__init__()
            self.target = target
            self.args = args
    
        def run(self):
            self.target(*self.args)
    
    
    if __name__ == "__main__":
        print(f'主线程开始时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')
    
        # 初始化3个线程，传递不同的参数
        t1 = MyThread(target=task_thread,args=(3,))
        t2 = MyThread(target=task_thread,args=(2,))
        t3 = MyThread(target=task_thread,args=(1,))
        # 开启三个线程
        t1.start()
        t2.start()
        t3.start()
        # 等待运行结束
        t1.join()
        t2.join()
        t3.join()
    
        print(f'主线程结束时间：{ time.strftime("%Y-%m-%d %H:%M:%S")}')

## 三. 多线程同步之lock（互斥锁） ##

不加锁的意外情况：  
3个线程对共同mun进行100万次加减操作之后，num的结果不为零

#!/usr/local/bin/python3
    #-*- coding: utf-8 -*-
    import time, threading
    
    num = 0
    
    def task_thread(n):
       global num
       for i in range(1000000):
           num = num + n
           num = num - n
    
    t1 = threading.Thread(target=task_thread, args=(6,))
    t2 = threading.Thread(target=task_thread, args=(17,))
    t3 = threading.Thread(target=task_thread, args=(11,))
    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()
    print(num)

加互斥锁之后的运行结果始终一致

import time, threading
    
    num = 0
    lock = threading.Lock()
    def task_thread(n):
        global num
        # 获取锁，用于线程同步
        lock.acquire()
        for i in range(1000000):
            num = num + n
            num = num - n
        #释放锁，开启下一个线程
        lock.release()
    
    t1 = threading.Thread(target=task_thread, args=(6,))
    t2 = threading.Thread(target=task_thread, args=(17,))
    t3 = threading.Thread(target=task_thread, args=(11,))
    t1.start()
    t2.start()
    t3.start()
    t1.join()
    t2.join()
    t3.join()
    print(num)

## 四. 多线程同步之Semaphore(信号量) ##

互斥锁只允许一个线程访问共享数据，而信号量同时允许一定数量的线程访问共享数据。

比如柜台有5个窗口，允许同时有5个人办理业务，后面的人只能等待，5人中有人办理完业务，等待的人才能去办理。  
使用信号量控制多线程并发数，代码如下：

import threading
    import time
    
    # 同时只有5个人办理业务
    semaphore = threading.BoundedSemaphore(5)
    
    
    # 模拟银行业务办理
    def yewubanli(name):
        semaphore.acquire()
        time.sleep(3)
        print(f"{ time.strftime('%Y-%m-%d %H:%M:%S')} { name} 正在办理业务")
    
        semaphore.release()
    
    
    thread_list = []
    for i in range(12):
        t = threading.Thread(target=yewubanli, args=(i,))
        thread_list.append(t)
    
    for thread in thread_list:
        thread.start()
    
    for thread in thread_list:
        thread.join()
    
    # while threading.active_count() != 1:
    # time.sleep(1)

可以看出，同一时刻只有5个人正在办理业务，即同一时刻只有5个线程获得资源运行。

## 五. 多线程同步之Condition ##

条件对象能让一个线程A停下来，等待其他线程B，线程B满足某个条件后通知线程B继续运行

import threading
    
    
    class Boy(threading.Thread):
        def __init__(self, cond, name):
            super(Boy, self).__init__()
            self.cond = cond
            self.name = name
    
        def run(self):
            self.cond.acquire()
            print(self.name + ": 嫁给我吧！？")
            self.cond.notify()  # 唤醒一个挂起的线程，让hanmeimei表态
            self.cond.wait()  # 释放内部所占用的琐，同时线程被挂起，直至接收到通知被唤醒或超时，等待hanmeimei回答
            print(self.name + ": 我单下跪，送上戒指！")
            self.cond.notify()
            self.cond.wait()
            print(self.name + ": Li太太，你的选择太明智了。")
            self.cond.release()
    
    
    class Girl(threading.Thread):
        def __init__(self, cond, name):
            super(Girl, self).__init__()
            self.cond = cond
            self.name = name
    
        def run(self):
            self.cond.acquire()
            self.cond.wait()  # 等待Lilei求婚
            print(self.name + ": 没有情调，不够浪漫，不答应")
            self.cond.notify()
            self.cond.wait()
            print(self.name + ": 好吧，答应你了")
            self.cond.notify()
            self.cond.release()
    
    
    cond = threading.Condition()
    boy = Boy(cond, "LiLei")
    girl = Girl(cond, "HanMeiMei")
    girl.start()
    boy.start()

上面程序先启动了girl线程，gitl虽然获取到了条件变量锁cond, 但又执行了wait并释放条件变量锁，自身进入阻塞状态。  
boy线程启动后，就获得了条件变量锁cond并发出了消息，之后通过notify唤醒一个挂起的线程。  
最后通过release程序释放资源。

## 六. 多线程同步之Event ##

事件用于线程之间的通信。一个线程发出一个信号，其他一个或者多个线程等待，调用Event对象的wait方法,线程则会阻塞等待，直到别的线程set之后才会被唤醒。

import threading, time
    
    
    class Boy(threading.Thread):
        def __init__(self, cond, name):
            super(Boy, self).__init__()
            self.cond = cond
            self.name = name
    
        def run(self):
            print(self.name + ": 嫁给我吧！？")
            self.cond.set()  # 唤醒一个挂起的线程，让hanmeimei表态
            time.sleep(0.5)
            self.cond.wait()
            print(self.name + ": 我单下跪，送上戒指！")
            self.cond.set()
            time.sleep(0.5)
            self.cond.wait()
            self.cond.clear()
            print(self.name + ": Li太太，你的选择太明智了。")
    
    
    class Girl(threading.Thread):
        def __init__(self, cond, name):
            super(Girl, self).__init__()
            self.cond = cond
            self.name = name
    
        def run(self):
            self.cond.wait()  # 等待Lilei求婚
            self.cond.clear()
            print(self.name + ": 没有情调，不够浪漫，不答应")
            self.cond.set()
            time.sleep(0.5)
            self.cond.wait()
            print(self.name + ": 好吧，答应你了")
            self.cond.set()
    
    
    cond = threading.Event()
    boy = Boy(cond, "LiLei")
    girl = Girl(cond, "HanMeiMei")
    boy.start()
    girl.start()

Event内部默认内置了一个标志，初始值为false。上面代码girl通过wait()方法进入等待状态，直到boy调用该Event的set()方法将内置标志设置为true，对象girl再继续运行。

对象boy最后调用Event的clear方法再讲内置的标志设置为False，恢复初始状态。

## 七. 线程优先队列 ##

python的queue模块提供了同步的，线程安全的队列类。包括

*  先进先出的队列queue;
 *  后进先出的LifoQueue;
 *  优先级队列PriorityQueue

这些队列都实现了锁原语，可以直接使用来实现线程的同步。

比如，有一个小冰箱用来存放冷饮，假如只能放5瓶冷饮，A不停的放，B不停的取，A和B的放取速度不一致，就可以用队列来做。

import threading, time
    
    import queue
    
    # 先进先出
    q = queue.Queue(maxsize=5)
    
    
    # q = queue.LifoQueue(maxsize=3)
    # q = queue.PriorityQueue(maxsize=3)
    
    def ProducerA():
        count = 1
        while True:
            q.put(f"冷饮 { count}")
            print(f"{ time.strftime('%H:%M:%S')} A 放入:[冷饮 { count}]")
            count += 1
            time.sleep(1)
            print(f'放入后队列长度：{ q.qsize()}')
    
    
    def ConsumerB():
        while True:
            print(f"{ time.strftime('%H:%M:%S')} B 取出 [{ q.get()}]")
            print('--', q.qsize())
            print(f'取出后队列长度：{ q.qsize()}')
            time.sleep(5)
    
    
    p = threading.Thread(target=ProducerA)
    c = threading.Thread(target=ConsumerB)
    c.start()
    p.start()

上面的代码就是实现生产者和消费者模型的一个比较简单的例子。

在并发编程中，使用生产者和消费之模式可以解决绝大多数的并发问题

如果生产者处理的速度很快，而消费者处理速度很慢，那么生产者就必须等消费者处理完，才能继续生产数据。  
同理，如果消费者的处理能力大于生产者，那消费者就必须等待生产者。

生产者和消费者模式就是通过一个容器（队列）来解决强耦合问题,生产者和消费者之间不通信，阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

## 八. 多线程之线程池 pool ##

单线程和线程池并发执行效率对比：

from multiprocessing.dummy import Pool as ThreadPool
    import time
    
    
    def fun(n):
        time.sleep(2)
    
    
    start = time.time()
    for i in range(5):
        fun(i)
    print("单线程顺序执行耗时:", time.time() - start)
    
    start2 = time.time()
    # 开8个 worker，没有参数时默认是 cpu 的核心数
    pool = ThreadPool(processes=5)
    # 在线程中执行 urllib2.urlopen(url) 并返回执行结果
    results2 = pool.map(fun, range(5))
    pool.close()
    pool.join()
    print("线程池（5）并发执行耗时:", time.time() - start2)

[python3_python3]: https://cuiyonghua.blog.csdn.net/article/details/119915966
[python3_python3 1]: https://cuiyonghua.blog.csdn.net/article/details/119917182
[python3 _python3]: https://cuiyonghua.blog.csdn.net/article/details/119917906