Python 编程 | 连载 26 - Python 多线程
Python 编程 | 连载 26 - Python 多线程
持续创作，加速成长！这是我参与「掘金日新计划 · 6 月更文挑战」的第1天，点击查看活动详情
一、多线程
线程是系统的最小调度单元，线程相比进程来说，对于资源的消耗低。线程可以通过threading模块下Thread函数来创建，线程对象的相关方法有：
- Thread：创建线程，入参需要传入函数名以及函数的参数，返回一个线程对象
- start：启动线程
- join：阻塞直到线程执行结束
- getName：获取线程名
- setName：设置线程名
- is_alive：判断线程是否存活
- setDaemon：守护线程
通过random.choice函数选中一个列中的元素，从列表中移除该元素并加入另外一个列表，直至列表为空。
import random, time
heros = ['stark', 'clint', 'thor', 'hulk', 'widow', 'captain', 'park', 'loki', 'strange', 'wanda']
_heros = []
def create():
if len(heros) == 0:
return
hero = random.choice(heros)
heros.remove(hero)
print('移除的元素为：{}'.format(hero))
_heros.append(hero)
time.sleep(1)
if __name__ == '__main__':
start = time.time()
for i in range(len(heros)):
create()
print('heros: {}'.format(heros))
print('_heros: {}'.format(_heros))
end = time.time()
print('耗时: {}'.format(end - start))
使用多线程方式处理，首先导入threading模块，修改main函数中的代码。
if __name__ == '__main__':
start = time.time()
threads = []
for i in range(len(heros)):
# 创建线程执行任务
thread = threading.Thread(target=create)
threads.append(thread)
thread.start()
print('线程名为：{}'.format(thread.getName()))
for thread in threads:
thread.join()
# create()
print('heros: {}'.format(heros))
print('_heros: {}'.format(_heros))
end = time.time()
print('耗时: {}'.format(end - start))
循环创建了10个线程，每个线程都去执行任务，整个耗时非常短。
通过线程执行任务存在的问题：
- 函数无法获取返回值
- 多个线程同时修改文件可能造成数据混乱
- 线程太多可能会造成资源不足
二、线程之间的通信
线程之间通信同样需要使用到队列。
import threading, time, queue, random
def send():
while True:
mes = random.randint(1, 10)
print('send message {}'.format(mes))
queue.put(mes)
time.sleep(1)
def receive():
while True:
x = queue.get()
print('receive {}'.format(x))
time.sleep(1)
if __name__ == '__main__':
queue = queue.Queue()
t1 = threading.Thread(target=send)
t2 = threading.Thread(target=receive)
t1.start()
t2.start()
t1.join()
t2.join()
线程池的创建
线程池可以避免线程创建和销毁带来的消耗，线程池需要通过futures.ThreadPoolExecutir方法来创建，线程池相关的方法有
- futures.ThreadPoolExecutor：创建线程池
- submit：往线程池中加入任务
- done：判断线程池中的某个线程是否完成任务
- result：获取线程执行的结果
首先导入concurrent.futures.thread包
import time, threading
from concurrent.futures.thread import ThreadPoolExecutor
def hallo(info):
print(info)
time.sleep(1)
if __name__ == '__main__':
# 创建线程池
pool = ThreadPoolExecutor(2)
for i in range(20):
pool.submit(hallo, ('Mark {}'.format(i)))
控制台输出结果时几乎是两条信息同时打印，这是因为此时有两个线程正在执行任务。
# 创建一个线程锁
lock = threading.Lock()
def hallo(info):
# 上锁
lock.acquire()
print(info)
time.sleep(1)
# 开锁
lock.release()
上了线程锁，就只能一个一个的执行了。
修改hallo()函数，返回info参数，并将上锁和解锁代码注释
# 其余代码不变
def hallo(info):
# 上锁
# lock.acquire()
# print(info)
time.sleep(1)
# 开锁
# lock.release()
# print('PID：{}'.format(os.getpid()))
return info
if __name__ == '__main__':
# hallo()函数返回结果的列表
results = []
# 创建线程池
pool = ThreadPoolExecutor(2)
for i in range(20):
res = pool.submit(hallo, ('Mark {}'.format(i)))
results.append(res)
print(res)
print(dir(res))
print('是否执行结束：{}'.format(res.done()))
# 遍历结果列表
for res in results:
print('遍历结果列表：{}'.format(res.result()))
print('是否执行结束：{}'.format(res.done()))
线程池通过submit提交一个任务执行，返回一个Future对象，可以从该对象中通过调用result()函数获取任务执行的返回值。
GIL全局锁
Python 解释器在执行的时候自动加的一把锁，造成Python中的多线程无法在多个core执行，只能在一个core上执行，这把锁就是GIL锁。GIL是全局解释器锁，并不是Python的特性，它是在Cpython解释器里引入的一个概念，而在其他语言编写的解释器里没有GIL。
GIL锁的作用：
- 单一CPU工作
- 确保线程安全
pypy解释器是没有GIL全局锁的，但是不推荐使用pypy解释器，推荐多进程+多线程的方式，通过多个进程在多个CPU上执行，每个进程在执行多个线程。
在CPython解释其中，当Python代码有一个线程开始访问解释器的时候，GIL就会给这个线程上锁，此时此刻线程只能等着，无法对解释器的资源进行访问，需要等待线程分配时间，这个线程把锁释放，另外的线程才开始运行。
三、异步
异步是相对于同步而言的，同步既指程序按照顺序一步一步往下执行，异步就是无序，无序等待上一步完成之后才可以执行下一步。
异步编程是一种并发编程的模式，其关注点是通过调度不同任务之间的执行和等待时间，通过减少处理器的闲置时间来达到减少整个程序的执行时间；异步编程跟同步编程模型最大的不同就是其任务的切换，当遇到一个需要等待长时间执行的任务的时候，我们可以切换到其他的任务执行。
asyncio 异步模块
async与await关键字：
- async：定义异步
- await：执行异步
相关函数：
- gather：将异步函数批量执行，返回一个列表，既函数执行结果的列表
- run：执行主异步函数，返回值是函数执行的返回值
import time, random
def zulu():
for i in range(10):
print('Zulu {}'.format(i))
time.sleep(random.random() * 2)
return 'zulu'
def tango():
for i in range(10):
print('Tango {}'.format(i))
time.sleep(random.random() * 2)
return 'tango'
if __name__ == '__main__':
start = time.time()
zulu()
tango()
end = time.time()
print('耗时:{}'.format(end-start))
将同步改为异步执行的方式
async def zulu():
for i in range(10):
print('Zulu {}'.format(i))
await asyncio.sleep(random.random() * 2)
return 'zulu'
async def tango():
for i in range(10):
print('Tango {}'.format(i))
await asyncio.sleep(random.random() * 2)
return 'tango'
async def main():
result = await asyncio.gather(
zulu(),
tango()
)
print(result)
if __name__ == '__main__':
start = time.time()
# zulu()
# tango()
asyncio.run(main())
end = time.time()
print('耗时:{}'.format(end-start))
zulu()和tango()两个函数交互执行，时间缩短一半
分别在zulu函数中和mian函数中打印出pid。
与多线程和多进程编程模型相比，异步编程只是在同一个线程之内的的任务调度
gevent 异步模块
gevent异步包需要通过pip进行安装
python3 -m pip install gevent -i https://pypi.tuna.tsinghua.edu.cn/simple
gevent 异步模块常用方法：
- spawn：创建协程对象，参数为func以及传入函数的参数，返回一个协程对象
- joinall：批量处理协程对象
- get：获取函数返回结果
- value：属性，也可以获取函数返回值
- join：阻塞等待异步程序结束
- kill：杀掉当前协程
- dead：判断当前协程是否销毁
import time, random, os
import gevent
def zulu_gevent():
for i in range(10):
print('Zulu Gevent, PID：{}'.format(os.getpid()))
gevent.sleep(random.random() ** 2)
return 'Zulu with Gevent'
def tango_gevent():
for i in range(10):
print('Tango Gevent, PID：{}'.format(os.getpid()))
gevent.sleep(random.random() ** 2)
return 'Tango with Gevent'
if __name__ == '__main__':
start = time.time()
zulu = gevent.spawn(zulu_gevent)
tango = gevent.spawn(tango_gevent)
res_gevent = [zulu, tango]
res = gevent.joinall(res_gevent)
print(res)
print(res[0].value)
print(res[1].get())
end = time.time()
print('耗时：{}'.format(end - start))
print('PID：{}'.format(os.getpid()))
调用value属性可以从协程对象中获取函数的返回值。
腾讯云开发者
