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

Posted 2021-09-11 cui_yonghua

1. python3多进程实战（python3经典编程案例）
2. python3多线程实战（python3经典编程案例）
3. 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以上是关于python3多线程实战（python3经典编程案例）的主要内容，如果未能解决你的问题，请参考以下文章
 
 python3多进程实战（python3经典编程案例）
 python3多进程实战（python3经典编程案例）
 python3 协程实战（python3经典编程案例）
 python3 协程实战（python3经典编程案例）
 python3分别用多进程，多线程，协程爬取豆瓣top250数据（python经典编程案例）
 用python的tushare模块分析股票案例（python3经典编程案例）