线程
一 threading模块介绍
multiprocess模块的完全模仿了threading模块的接口,二者在使用层面,有很大的相似性,因而不再详细介绍
二 开启线程的两种方式
方式一
from threading import Thread
import time
def sayhi(name):
time.sleep(2)
print('%s say hello' %name)
if __name__ == '__main__':
t=Thread(target=sayhi,args=('egon',))
t.start()
print('主线程')
方式二
#方式二
from threading import Thread
import time
class Sayhi(Thread):
def __init__(self,name):
super().__init__()
self.name=name
def run(self):
time.sleep(2)
print('%s say hello' % self.name)
if __name__ == '__main__':
t = Sayhi('egon')
t.start()
print('主线程')
练习题
1、基于多线程实现并发的套接字通信
server
from threading import Thread
import socket
IP_PORT = ('127.0.0.1', 8080)
def server1(IP_PORT):
server = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server.bind(IP_PORT)
server.listen(5)
while True:
conn, client_addr = server.accept()
t = Thread(target=talk, args=(conn,))
t.start()
server1.closer()
def talk(conn):
while True:
try:
msg = conn.recv(1024)
if not msg:
continue
except ConnectionResetError as e:
print(e)
break
conn.send(msg.upper())
if __name__ == '__main__':
server1(IP_PORT)
client
import socket
IP_PORT = ('127.0.0.1', 8080)
client = socket.socket(family=socket.AF_INET, type=socket.SOCK_STREAM)
client.connect(IP_PORT)
while True:
msg = input('>>>:').strip()
if not msg:
continue
client.send(msg.encode('utf-8'))
data_recv = client.recv(1024)
print(data_recv.decode('utf-8'))
三 多进程与多线程的区别
谁的开启速度快
- 在主进程下开启线程
- 在主进程下开启子进程
进程是CPU的最小资源分配单位,线程是CPU的最小可支配的单位.
PID
- 在主进程下开启多个线程,每个线程都跟主进程的pid一样
- 开启多个进程,每个进程都有不同的pid
同一进程内的线程共享该进程的地址空间
- 进程之间地址空间的隔离的
- 同一进程内的多个线程是共享该进程地址空间的
四 Thread对象的其他属性或方法
介绍
Thread实例对象的方法
isAlive(): 返回线程是否活动的。
getName(): 返回线程名。
setName(): 设置线程名。
threading模块提供的一些方法:
threading.currentThread(): 返回当前的线程变量。
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前,不包括启动前和终止后的线程。
threading.activeCount(): 返回正在运行的线程数量,与len(threading.enumerate())有相同的结果。
验证
from threading import Thread
import threading
from multiprocessing import Process
import os
def work():
import time
time.sleep(3)
print(threading.current_thread().getName())
if __name__ == '__main__':
#在主进程下开启线程
t=Thread(target=work)
t.start()
print(threading.current_thread().getName())
print(threading.current_thread()) #主线程
print(threading.enumerate()) #连同主线程在内有两个运行的线程
print(threading.active_count())
print('主线程/主进程')
执行结果
MainThread
<_MainThread(MainThread, started 140735268892672)>
[<_MainThread(MainThread, started 140735268892672)>, <Thread(Thread-1, started 123145307557888)>]
主线程/主进程
Thread-1
五 线程守护
- 守护线程是对主线程的守护,而主线程会等到其他所有非守护线程结束之后,自己才会结束,因为主线程代表了整个进程的生命周期.所以守护线程的结果就是等到当前进程结束以后,守护线程才会结束
from threading import Thread
import time
def foo():
print(123)
time.sleep(1)
print("end123")
def bar():
print(456)
time.sleep(3)
print("end456")
if __name__ == '__main__':
t1=Thread(target=foo)
t2=Thread(target=bar)
t1.daemon=True
t1.start()
t2.start()
print("main-------")
六 GIL全局解释器锁
前言:
- 在Cpython解释器中,同一个进程下开启的多线程,同一时刻只能有一个线程执行,无法利用多核优势
- GIL并不是Python的特性,它是在实现Python解析器(CPython)时所引入的一个概念。GIL并不是Python的特性,Python完全可以不依赖于GIL
GIL介绍
-
GIL本质就是一把互斥锁,既然是互斥锁,所有互斥锁的本质都一样,都是将并发运行变成串行,以此来控制同一时间内共享数据只能被一个任务所修改,进而保证数据安全。
可以肯定的一点是:保护不同的数据的安全,就应该加不同的锁。
要想了解GIL,首先确定一点:每次执行python程序,都会产生一个独立的进程。例如python test.py,python aaa.py,python bbb.py会产生3个不同的python进程
1、所有数据都是共享的,
这其中,代码作为一种数据也是被所有线程共享的
(test.py的所有代码以及Cpython解释器的所有代码)
例如:test.py定义一个函数work(代码内容如下图),
在进程内所有线程都能访问到work的代码,
于是我们可以开启三个线程然后target都指向该代码,
能访问到意味着就是可以执行。
2、所有线程的任务,
都需要将任务的代码当做参数传给解释器的代码去执行,
即所有的线程要想运行自己的任务,
首先需要解决的是能够访问到解释器的代码。
GIL与Lock
GIL 与Lock是两把锁,保护的数据不一样,
前者是解释器级别的(当然保护的就是解释器级别的数据,比如垃圾回收的数据),
后者是保护用户自己开发的应用程序的数据,
很明显GIL不负责这件事,只能用户自定义加锁处理,即Lock
GIL与多线程
GIL的存在,同一时刻同一进程中只有一个线程被执行
应用:
多线程用于IO密集型,如socket,爬虫,web
多进程用于计算密集型,如金融分析
七 死锁现象与递归锁
一 死锁现象
所谓死锁: 是指两个或两个以上的进程或线程在执行过程中,因争夺资源而造成的一种互相等待的现象,若无外力作用,它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁,这些永远在互相等待的进程称为死锁进程,如下就是死锁
from threading import Thread,Lock
import time
mutexA=Lock()
mutexB=Lock()
class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print('\033[41m%s 拿到A锁\033[0m' %self.name)
mutexB.acquire()
print('\033[42m%s 拿到B锁\033[0m' %self.name)
mutexB.release()
mutexA.release()
def func2(self):
mutexB.acquire()
print('\033[43m%s 拿到B锁\033[0m' %self.name)
time.sleep(2)
mutexA.acquire()
print('\033[44m%s 拿到A锁\033[0m' %self.name)
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t=MyThread()
t.start()
执行效果
Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-2 拿到A锁 #出现死锁,整个程序阻塞住
二 递归锁
解决方法,递归锁,在Python中为了支持在同一线程中多次请求同一资源,python提供了可重入锁RLock。
这个RLock内部维护着一个Lock和一个counter变量,counter记录了acquire的次数,从而使得资源可以被多次require。直到一个线程所有的acquire都被release,其他的线程才能获得资源。上面的例子如果使用RLock代替Lock,则不会发生死锁,二者的区别是:递归锁可以连续acquire多次,而互斥锁只能acquire一次
在RLock()中,每acquire()一次,计数器就加一,每release()一次,计数器就减一,只有计数器为0的时候,其他的线程才能抢到acquire(),保证了不会死锁.
from threading import Thread,RLock
import time
mutexA=mutexB=RLock() #一个线程拿到锁,counter加1,该线程内又碰到加锁的情况,则counter继续加1,这期间所有其他线程都只能等待,等待该线程释放所有锁,即counter递减到0为止
class MyThread(Thread):
def run(self):
self.func1()
self.func2()
def func1(self):
mutexA.acquire()
print('\033[41m%s 拿到A锁\033[0m' %self.name)
mutexB.acquire()
print('\033[42m%s 拿到B锁\033[0m' %self.name)
mutexB.release()
mutexA.release()
def func2(self):
mutexB.acquire()
print('\033[43m%s 拿到B锁\033[0m' %self.name)
time.sleep(2)
mutexA.acquire()
print('\033[44m%s 拿到A锁\033[0m' %self.name)
mutexA.release()
mutexB.release()
if __name__ == '__main__':
for i in range(10):
t=MyThread()
t.start()
八 信号量,Event,定时器
信号量
信号量也是一把锁,可以指定信号量为5,对比互斥锁同一时间只能有一个任务抢到锁去执行,信号量同一时间可以有5个任务拿到锁去执行,如果说互斥锁是合租房屋的人去抢一个厕所,那么信号量就相当于一群路人争抢公共厕所,公共厕所有多个坑位,这意味着同一时间可以有多个人上公共厕所,但公共厕所容纳的人数是一定的,这便是信号量的大小
from threading import Thread,Semaphore
import threading
import time
# def func():
# sm.acquire()
# print('%s get sm' %threading.current_thread().getName())
# time.sleep(3)
# sm.release()
with sm: # 简写方法,自动在开头结尾加acquire(),release()
print('%s get sm' %threading.current_thread().getName())
time.sleep(3)
if __name__ == '__main__':
sm=Semaphore(5)
for i in range(23):
t=Thread(target=func)
t.start()
解析
Semaphore管理一个内置的计数器,
每当调用acquire()时内置计数器-1;
调用release() 时内置计数器+1;
计数器不能小于0;当计数器为0时,acquire()将阻塞线程直到其他线程调用release()。
Event
线程的一个关键特性是每个线程都是独立运行且状态不可预测。如果程序中的其 他线程需要通过判断某个线程的状态来确定自己下一步的操作,这时线程同步问题就会变得非常棘手。为了解决这些问题,我们需要使用threading库中的Event对象。 对象包含一个可由线程设置的信号标志,它允许线程等待某些事件的发生。在 初始情况下,Event对象中的信号标志被设置为假。如果有线程等待一个Event对象, 而这个Event对象的标志为假,那么这个线程将会被一直阻塞直至该标志为真。一个线程如果将一个Event对象的信号标志设置为真,它将唤醒所有等待这个Event对象的线程。如果一个线程等待一个已经被设置为真的Event对象,那么它将忽略这个事件, 继续执行
from threading import Event
event.isSet():返回event的状态值;
event.wait():如果 event.isSet()==False将阻塞线程;
event.set(): 设置event的状态值为True,所有阻塞池的线程激活进入就绪状态, 等待操作系统调度;
event.clear():恢复event的状态值为False。
例如,有多个工作线程尝试链接MySQL,我们想要在链接前确保MySQL服务正常才让那些工作线程去连接MySQL服务器,如果连接不成功,都会去尝试重新连接。那么我们就可以采用threading.Event机制来协调各个工作线程的连接操作
from threading import Thread,Event
import threading
import time,random
def conn_mysql():
count=1
while not event.is_set():
if count > 3:
raise TimeoutError('链接超时')
print('<%s>第%s次尝试链接' % (threading.current_thread().getName(), count))
event.wait(0.5)
count+=1
print('<%s>链接成功' %threading.current_thread().getName())
def check_mysql():
print('\033[45m[%s]正在检查mysql\033[0m' % threading.current_thread().getName())
time.sleep(random.randint(2,4))
event.set()
if __name__ == '__main__':
event=Event()
conn1=Thread(target=conn_mysql)
conn2=Thread(target=conn_mysql)
check=Thread(target=check_mysql)
conn1.start()
conn2.start()
check.start()
定时器
定时器,指定n秒后执行某操作
from threading import Timer
def hello():
print("hello, world")
t = Timer(1, hello)
t.start() # after 1 seconds, "hello, world" will be printed
九 线程queue
queue is especially useful in threaded programming when information must be exchanged safely between multiple threads.
有三种不同的用法
class queue.Queue(maxsize=0) #队列:先进先出
import queue
q=queue.Queue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
'''
结果(先进先出):
first
second
third
'''
class queue.LifoQueue(maxsize=0) #堆栈:last in fisrt out
import queue
q=queue.LifoQueue()
q.put('first')
q.put('second')
q.put('third')
print(q.get())
print(q.get())
print(q.get())
'''
结果(后进先出):
third
second
first
'''
class queue.PriorityQueue(maxsize=0) #优先级队列:存储数据时可设置优先级的队列
import queue
q=queue.PriorityQueue()
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高
q.put((20,'a'))
q.put((10,'b'))
q.put((30,'c'))
print(q.get())
print(q.get())
print(q.get())
'''
结果(数字越小优先级越高,优先级高的优先出队):
(10, 'b')
(20, 'a')
(30, 'c')
'''
十 进程池与线程池
介绍
官网:https://docs.python.org/dev/library/concurrent.futures.html
concurrent.futures模块提供了高度封装的异步调用接口
ThreadPoolExecutor:线程池,提供异步调用
ProcessPoolExecutor: 进程池,提供异步调用
Both implement the same interface, which is defined by the abstract Executor class.
基本方法
1、submit(fn, *args, **kwargs)
异步提交任务
2、map(func, *iterables, timeout=None, chunksize=1)
取代for循环submit的操作
3、shutdown(wait=True)
相当于进程池的pool.close()+pool.join()操作
wait=True,等待池内所有任务执行完毕回收完资源后才继续
wait=False,立即返回,并不会等待池内的任务执行完毕
但不管wait参数为何值,整个程序都会等到所有任务执行完毕
submit和map必须在shutdown之前
4、result(timeout=None)
取得结果
5、add_done_callback(fn)
回调函数
十一 异步调用与回调机制
- 同步调用:
- 提交完任务后,原地等待该任务执行完毕,拿到结果,再执行下一行代码,导致程序是串行执行.
-
rsponse = pool.submit(func1).result()
这里.result()
的作用是拿到func1
的return
值 - 然后执行
func2(response)
来处理结果 - 这样多个进程的话就成了串行执行.
- 异步调用:
- 提交完任务后,不再原地等待任务执行完毕
pool.submit(func1).add_done_callback(func2)
- 自动将提交的任务
func1
执行的结果作为一个对象参数传给.add_done_callback(func2)
中的func2
,这里func2
只接受一个参数,然后在func2
中首先执行response = obj.result()
拿到func1
的return
值,来进行下一步处理.
from concurrent.futures import ThreadPoolExecutor,ProcessPoolExecutor
from multiprocessing import Pool
import requests
import json
import os
def get_page(url):
print('<进程%s> get %s' %(os.getpid(),url))
respone=requests.get(url)
if respone.status_code == 200:
return {'url':url,'text':respone.text}
def parse_page(res):
res=res.result()
print('<进程%s> parse %s' %(os.getpid(),res['url']))
parse_res='url:<%s> size:[%s]\n' %(res['url'],len(res['text']))
with open('db.txt','a') as f:
f.write(parse_res)
if __name__ == '__main__':
urls=[
'https://www.baidu.com',
'https://www.python.org',
'https://www.openstack.org',
'https://help.github.com/',
'http://www.sina.com.cn/'
]
p=ProcessPoolExecutor(3)
for url in urls:
p.submit(get_page,url).add_done_callback(parse_page) #parse_page拿到的是一个future对象obj,需要用obj.result()拿到结果