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)