python基础之线程、进程、协程

线程

线程基础知识

一个应用程序，可以多进程、也可以多线程.

一个python脚本，默认是单进程，单线程的。

I/O操作（音频、视频、显卡操作），不占用CPU，所以：

对于I/O密集型操作，不会占用CPU，使用多线程操作，能提高效率

对于计算密集型操作，由于占用CPU，使用多进程操作，能提高效率

python中有个全局解释器锁，叫GIL（全称

Global Interpreter Lock

），导致一个进程只能由一个线程让CPU去调度,但在java c#可以使用多个线程。

多线程，多进程的目的，是为了提高并发，I/O密集型用多线程，计算密集型，用多进程。

我们来看看怎么创建多线程：

def f1(args):
print(args)
import threading
t=threading.Thread(target=f1,args=(123,))    #创建一个线程，target表示线程执行的目标，args表示参数
t.start()     #并不代表当前立即被执行,系统来决定
f1(111)

以上代码结果print顺序会随机！

更多的方法：

start 不代表当前线程并不会立即被执行，而是等待CPU调度,(准备就绪,等待调度)

setName 为线程设置名称

setDaemon(True) True表示主线程不等待子线程,执行完自己的任务后，自动关闭，子线程有可能未执行完毕。（默认情况下，主线程要等待子线程执行完毕后再关闭主线程）,(True:后台线程，主线程执行过程中，后台线程也在进行，主线程执行完毕后，后台线程不论成功与否，均停止;False:前台线程，主线程执行过程中，前台线程也在进行，主线程执行完毕后，等待前台线程也执行完成后，程序停止)

join(2) 如果不想让线程并发的操作，表示主线程到此等待，等待直到子线程执行完毕。如果加上参数，表示主线程在此最多等几秒。该方法使得多线程变得无意义

run 线程被cpu调度后自动执行线程对象的run方法

import time

def f1(args):
time.sleep(5)
print(args)

import threading
t1=threading.Thread(target=f1,args=(123,))
t1.setDaemon(True)  #表示主线程不等待子线程
t.start()     #并不代表当前被立即被执行,系统来决定
f1(111)

t.join(2) #表示主程序执行到此,等待...直到子线程执行完毕
print(222222)
print(333333)

下面看下run方法:

class MyThread(threading.Thread):
def __init__(self,num):
threading.Thread.__init__(self)
self.num = num

def run(self):#定义每个线程要运行的函数

print("running on number:%s" %self.num)

time.sleep(3)

if __name__ == '__main__':

t1 = MyThread(1)
t2 = MyThread(2)
t1.start()
t2.start()

线程的调用方式:

线程有两种调用方式:

简单的调用方式

自定义类的调用方式

简单调用方式:自定义一个继承threading.thread的子类,通过自定义类的对象调用

import threading

def f1(arg):
print(arg)

t = threading.Thread(target=f1,args=(123,))
t.run()

自定义类的调用方式

import threading

def f2(arg):
print(arg)

class MyThread(threading.Thread):
def __init__(self,func,args):
self.func=func
self.args=args
super(MyThread,self).__init__()

def run(self):
self.func(self.args)

obj=MyThread(f2,123)
obj.run()

线程锁

由于线程之间进行随机调度,并且每个线程可能只执行n条操作后,当多个线程同时修改同一条数据时,可能会出现脏数据:同一时刻只能允许指定的线程数执行操作.

python中的线程锁有

Lock

,

RLock

两种,其中RLock用的较多,因为支持多层嵌套的方式,Lock用的较少,不支持多层嵌套锁.

def func(l):
global NUM
#上锁
l.acquire()
NUM-=1
time.sleep(2)
print(NUM)
#开锁
l.release()

lock=threading.RLock() #放行几个线程出去执行

for i in range(30):
t=threading.Thread(target=func,args=(lock,))
t.start()

如果不使用线程锁,上面程序会有30个线程同时执行,结果为30个-20

信号量(互斥锁)

semaphore,同时允许指定数量的线程更改数据,比如厕所有3个坑，那最多只允许3个人上厕所，后面的人只能等里面有人出来了才能再进去。

import threading,time

NUM=10

def func(l):
global NUM
#上锁
l.acquire()
NUM-=1
time.sleep(2)
print(NUM)
#开锁
l.release()

lock=threading.BoundedSemaphore(3)  #放行几个线程出去执行

for i in range(30):
t=threading.Thread(target=func,args=(lock,))
t.start()

从上面两个代码对比,我们会发现,semaphore如果设置为1时,也可实现信号锁的功能.

事件(event)

python线程中的event主要用于让主线程控制其子线程的执行方式(有点类似交警控制红绿灯),event主要提供三个方法:set wait clear

事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法时就会阻塞，如果“Flag”值为True，那么event.wait 方法时便不再阻塞。

clear:将'Flag'设置为False

set:将'Flag'设置为True

import threading

def func(i,e):
print(i)
e.wait()        #检测是什么灯
print(i+100)

event=threading.Event()
for i in range(10):
t = threading.Thread(target=func,args=(i,event))
t.start()

event.clear() #默认是红灯
inp = input('>>>')
if inp == '1':
event.set() #设置成绿灯

条件(condition)

使线程等待,当条件成立时,释放线程执行.

import threading

def func(i,con):
print(i)
con.acquire()   #配合,固定格式,线程hold住
con.wait()
print(i+100)
con.release()

c=threading.Condition()
for i in range(10):
t = threading.Thread(target=func,args=(i,c,))
t.start()

while True:
inp = input('>>>|')
if inp == 'q':
break
c.acquire()     #以下都是固定格式
c.notify(int(inp))
c.release()

例子1中,写到函数func中的:

c.acquire(),c.notify(args),c.release()

是固定格式.

例子2:wait_for

import threading

def condition():
ret=False
r = input('>>|')
if r == 'true':
ret= True
else:
ret=False
return ret

def func(i,con):
print(i)
con.acquire()   #配合,固定格式,线程hold住
con.wait_for(condition)
print(i+100)
con.release()

c=threading.Condition()
for i in range(10):
t = threading.Thread(target=func,args=(i,c,))
t.start()

其中例子2中,

con.acquire(),con.wait_for(condition)

是固定格式配合使用,拦截线程,con.release()释放线程.

定时器(timer):

定时器,延迟多长时间(单位:秒)执行

import threading

def hello():
print('hello,world!!')

t=threading.Timer(1,hello)
t.start()

线程池

python的线程池有两种实现方式,我们先来看一个比较简单的实现方式.

实现思路:

通过队列(先进先出队列,队列都是在内存中操作,进程退出,队列清空)来实现,线程池中没有线程时为阻塞状态.

自定义一个线程池类,构造方法时,创建一个指定元素数量的队列,队列中的元素为

线程类

使用线程时,使用队列的get方法得到一个线程类,使用__call__方法创建一个线程,使用线程执行指定的程序

程序执行完成后,在队列中添加一个新的

线程类

import threading,time,queue

class ThreadPool:
def __init__(self,maxsize):
self.maxsize=maxsize
self._q=queue.Queue(maxsize)
for i in range(maxsize):
self._q.put(threading.Thread)

def get_thread(self):
return self._q.get()
def add_thread(self):
self._q.put(threading.Thread)

pool=ThreadPool(5)

def task(arg,p):
print(arg)
time.sleep(1)
p.add_thread()

for i in range(100):
t = pool.get_thread()   #线程池中没有线程为阻塞状态
obj=t(target=task,args=(i,pool))
obj.start()

此方式的缺点:没有将线程重复利用,要直到创建一个线程的耗时可能是一个线程执行的好几倍,所以有了第二种方式.

第二种方式是也是使用队列,但队列中的元素为为一个个

(函数名,函数参数,)

的元组,创建一个线程组成的列表,线程轮流去队列中取到元组,分解后执行函数,然后取下一个函数.

import queue
import threading
import contextlib
import time

StopEvent = object()

class ThreadPool(object):

def __init__(self, max_num, max_task_num = None):
if max_task_num:
self.q = queue.Queue(max_task_num)
else:
self.q = queue.Queue()
self.max_num = max_num
self.cancel = False
self.terminal = False
self.generate_list = []
self.free_list = []

def run(self, func, args, callback=None):
"""
线程池执行一个任务
:param func: 任务函数
:param args: 任务函数所需参数
:param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数1、任务函数执行状态；2、任务函数返回值（默认为None，即：不执行回调函数）
:return: 如果线程池已经终止，则返回True否则None
"""
if self.cancel:
return
if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
self.generate_thread()
w = (func, args, callback,)
self.q.put(w)

def generate_thread(self):
"""
创建一个线程
"""
t = threading.Thread(target=self.call)
t.start()

def call(self):
"""
循环去获取任务函数并执行任务函数
"""
current_thread = threading.currentThread
self.generate_list.append(current_thread)

event = self.q.get()
while event != StopEvent:

func, arguments, callback = event
try:
result = func(*arguments)
success = True
except Exception as e:
success = False
result = None

if callback is not None:
try:
callback(success, result)
except Exception as e:
pass

with self.worker_state(self.free_list, current_thread):
if self.terminal:
event = StopEvent
else:
event = self.q.get()
else:

self.generate_list.remove(current_thread)

def close(self):
"""
执行完所有的任务后，所有线程停止
"""
self.cancel = True
full_size = len(self.generate_list)
while full_size:
self.q.put(StopEvent)
full_size -= 1

def terminate(self):
"""
无论是否还有任务，终止线程
"""
self.terminal = True

while self.generate_list:
self.q.put(StopEvent)

self.q.empty()

@contextlib.contextmanager
def worker_state(self, state_list, worker_thread):
"""
用于记录线程中正在等待的线程数
"""
state_list.append(worker_thread)
try:
yield
finally:
state_list.remove(worker_thread)

# How to use

pool = ThreadPool(5)

def callback(status, result):
# status, execute action status
# result, execute action return value
pass

def action(i):
print(i)

for i in range(30):
ret = pool.run(action, (i,), callback)

time.sleep(5)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
pool.close()
pool.terminate()

进程(multiprocessing)

进程与线程的使用方式基本雷同.比如start,daemon(用法略不同,意义相同),join,各种锁等等.

进程之间的数据共享

默认进程之间是无法进行共享的,看例子:

from multiprocessing import Process

li = []

def foo(i):
li.append(i)
print('say hi',li)

for i in range(10):
p = Process(target=foo,args=(i,))
p.start()

print('ending',li)

out:

say hi [0]
say hi [1]
say hi [2]
say hi [3]
say hi [4]
say hi [5]
say hi [6]
say hi [7]
ending []
say hi [8]
say hi [9]

那么如何让进程之间能够共享呢?

基本可分为三种方式:

from multiprocessing import queues 特殊的queues

from multiprocessing import Array 数组方式

from multiprocessing import Manager manager.dict

需要注意的是,他们都是multiprocessing中的模块

queues方式:

from multiprocessing import queues
from multiprocessing import Process
import multiprocessing
def foo(i,arg):
arg.put(i)
print('say hi',i,arg.qsize())

if __name__ == '__main__':
li=queues.Queue(20,ctx=multiprocessing)
for i in range(10):
p=Process(target=foo,args=(i,li,))
p.start()

Array方式,数组有个特性,必须初始化的时候指定数组的长度和元素类型:

from multiprocessing import Process
from multiprocessing import Array
def foo(i,arg):
arg[i]=i+100
for item in arg:
print(item)
print('======')

if __name__ == '__main__':
li=Array('i',10)
for i in range(10):
p=Process(target=foo,args=(i,li,))
p.start()

Array的类型对应表:

'c': ctypes.c_char,  'u': ctypes.c_wchar,
'b': ctypes.c_byte,  'B': ctypes.c_ubyte,
'h': ctypes.c_short, 'H': ctypes.c_ushort,
'i': ctypes.c_int,   'I': ctypes.c_uint,
'l': ctypes.c_long,  'L': ctypes.c_ulong,
'f': ctypes.c_float, 'd': ctypes.c_double

manager.dict 可实现数据共享

进程和进程之间如果想通讯,需要连接p.join()

from multiprocessing import Process
from multiprocessing import Manager
def foo(i,arg):
arg[i]=i+100
print(arg.values())
if __name__ == '__main__':
obj=Manager()
li=obj.dict()
for i in range(10):
p=Process(target=foo,args=(i,li,))
p.start()
p.join()

当创建进程时（非使用时），共享数据会被拿到子进程中，当进程中执行完毕后，再赋值给原值。

进程锁例子:

from multiprocessing import Process, Array, RLock

def Foo(lock,temp,i):
"""
将第0个数加100
"""
lock.acquire()
temp[0] = 100+i
for item in temp:
print(i,'----->',item)
lock.release()

lock = RLock()
temp = Array('i', [11, 22, 33, 44])

for i in range(20):
p = Process(target=Foo,args=(lock,temp,i,))
p.start()

进程池

进程池内部维护一个进程序列，当使用时，则去进程池中获取一个进程，如果进程池序列中没有可供使用的进程，那么程序就会等待，直到进程池中有可用进程为止。

进程池有两种方式:

apply 串行操作

异步操作apply_async

串行操作:

from multiprocessing import Pool
import time

def f1(arg):
time.sleep(1)
print(arg)
if __name__ == '__main__':
pool=Pool(5)
for i in range(30):
pool.apply(func=f1,args=(i,))

异步操作:

from multiprocessing import Pool
import time

def f1(arg):
time.sleep(1)
print(arg)
if __name__ == '__main__':
pool=Pool(5)
for i in range(30):
pool.apply_async(func=f1,args=(i,))
pool.close() #所有任务执行完毕          #1
time.sleep(1)
#pool.terminate() #立即终止,不管是否有任务正在执行或者待执行     #2
pool.join()

其中 #1 #2 二选一操作

pool.terminate 立即终止,不管是否有任务正在执行或者等待执行

pool.close 所有任务执行完毕后关闭

协程

原理:利用一个线程,分解一个线程成为多个微线程==>程序级别做的,与操作系统没有关系.

与线程进程的区别:

线程和进程的操作是由程序触发系统接口，最后的执行者是系统；协程的操作则是程序员。

协程的适用场景:涉及到http的I/O请求,协程是高性能的代名词.所以,网络爬虫很多是使用协程方式.

协程存在的意义：对于多线程应用，CPU通过切片的方式来切换线程间的执行，线程切换时需要耗时（保存状态，下次继续）。协程，则只使用一个线程，在一个线程中规定某个代码块执行顺序。

协程的执行方式:打个比方,1个很牛逼的足球队员,前面一排并列的足球,从第一个足球踢出去,然后提出第二个第三个,等足球弹回起始位置时,足球员对此足球接住后再次剃出或者停住球,这个足球员就是协程

协程使用前提

使用前需要安装gevent第三方模块

pip3 install gevent

看下代码吧,自动切换,关键词

gevent.spawn()

:

from gevent import monkey;monkey.patch_all()
import gevent
import requests

def f(url):
print('GET: %s' % url)
resp = requests.get(url)
data = resp.text
print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([
gevent.spawn(f, 'https://www.python.org/'),
gevent.spawn(f, 'https://www.yahoo.com/'),
gevent.spawn(f, 'https://github.com/'),
])