python语言由psf组织所有_python线程、进程和协程

Python被人诟病最多的大概就是性能差&＃xff0c;在这里讲一下 Python 的多进程&＃xff0c;多线程与协程。首先声明这不是教程&＃xff0c;看完这篇文章&＃xff0c;大概能够对 Python 的多进程与多线程有一定的了解。

引言

解释器环境&＃xff1a;python3.5.1

我们都知道python网络编程的两大必学模块socket和socketserver&＃xff0c;其中的socketserver是一个支持IO多路复用和多线程、多进程的模块。一般我们在socketserver服务端代码中都会写这么一句&＃xff1a;

server &＃61; socketserver.ThreadingTCPServer(settings.IP_PORT, MyServer)

ThreadingTCPServer这个类是一个支持多线程和TCP协议的socketserver&＃xff0c;它的继承关系是这样的&＃xff1a;

class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass

右边的TCPServer实际上是它主要的功能父类&＃xff0c;而左边的ThreadingMixIn则是实现了多线程的类&＃xff0c;它自己本身则没有任何代码。

MixIn在python的类命名中&＃xff0c;很常见&＃xff0c;一般被称为“混入”&＃xff0c;戏称“乱入”&＃xff0c;通常为了某种重要功能被子类继承。

class ThreadingMixIn:

daemon_threads &＃61; False

def process_request_thread(self, request, client_address):

try:

self.finish_request(request, client_address)

self.shutdown_request(request)

except:

self.handle_error(request, client_address)

self.shutdown_request(request)

def process_request(self, request, client_address):

t &＃61; threading.Thread(target &＃61; self.process_request_thread,

args &＃61; (request, client_address))

t.daemon &＃61; self.daemon_threads

t.start()

在ThreadingMixIn类中&＃xff0c;其实就定义了一个属性&＃xff0c;两个方法。在process_request方法中实际调用的正是python内置的多线程模块threading。这个模块是python中所有多线程的基础&＃xff0c;socketserver本质上也是利用了这个模块。

一、线程

代码如下:

线程&＃xff0c;有时被称为轻量级进程(Lightweight Process&＃xff0c;LWP)&＃xff0c;是程序执行流的最小单元。一个标准的线程由线程ID&＃xff0c;当前指令指针(PC)&＃xff0c;寄存器集合和堆栈组成。另外&＃xff0c;线程是进程中的一个实体&＃xff0c;是被系统独立调度和分派的基本单位&＃xff0c;线程自己不独立拥有系统资源&＃xff0c;但它可与同属一个进程的其它线程共享该进程所拥有的全部资源。一个线程可以创建和撤消另一个线程&＃xff0c;同一进程中的多个线程之间可以并发执行。由于线程之间的相互制约&＃xff0c;致使线程在运行中呈现出间断性。线程也有就绪、阻塞和运行三种基本状态。就绪状态是指线程具备运行的所有条件&＃xff0c;逻辑上可以运行&＃xff0c;在等待处理机&＃xff1b;运行状态是指线程占有处理机正在运行&＃xff1b;阻塞状态是指线程在等待一个事件(如某个信号量)&＃xff0c;逻辑上不可执行。每一个应用程序都至少有一个进程和一个线程。线程是程序中一个单一的顺序控制流程。在单个程序中同时运行多个线程完成不同的被划分成一块一块的工作&＃xff0c;称为多线程。

以上那一段&＃xff0c;可以不用看&＃xff01;举个例子&＃xff0c;厂家要生产某个产品&＃xff0c;在它的生产基地建设了很多厂房&＃xff0c;每个厂房内又有多条流水生产线。所有厂房配合将整个产品生产出来&＃xff0c;某个厂房内的所有流水线将这个厂房负责的产品部分生产出来。每个厂房拥有自己的材料库&＃xff0c;厂房内的生产线共享这些材料。而每一个厂家要实现生产必须拥有至少一个厂房一条生产线。那么这个厂家就是某个应用程序&＃xff1b;每个厂房就是一个进程&＃xff1b;每条生产线都是一个线程。

1.1 普通的多线程

在python中&＃xff0c;threading模块提供线程的功能。通过它&＃xff0c;我们可以轻易的在进程中创建多个线程。下面是个例子&＃xff1a;

import threading

import time

def show(arg):

time.sleep(1)

print(&＃39;thread&＃39;&＃43;str(arg))

for i in range(10):

t &＃61; threading.Thread(target&＃61;show, args&＃61;(i,))

t.start()

print(&＃39;main thread stop&＃39;)

上述代码创建了10个“前台”线程&＃xff0c;然后控制器就交给了CPU&＃xff0c;CPU根据指定算法进行调度&＃xff0c;分片执行指令。

下面是Thread类的主要方法&＃xff1a;

start 线程准备就绪&＃xff0c;等待CPU调度

setName 为线程设置名称

getName 获取线程名称

setDaemon 设置为后台线程或前台线程(默认)

如果是后台线程&＃xff0c;主线程执行过程中&＃xff0c;后台线程也在进行&＃xff0c;主线程执行完毕后&＃xff0c;后台线程不论成功与否&＃xff0c;均停止。如果是前台线程&＃xff0c;主线程执行过程中&＃xff0c;前台线程也在进行&＃xff0c;主线程执行完毕后&＃xff0c;等待前台线程也执行完成后&＃xff0c;程序停止。

join 逐个执行每个线程&＃xff0c;执行完毕后继续往下执行&＃xff0c;该方法使得多线程变得无意义。

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

1.2 自定义线程类

对于threading模块中的Thread类&＃xff0c;本质上是执行了它的run方法。因此可以自定义线程类&＃xff0c;让它继承Thread类&＃xff0c;然后重写run方法。

import threading

class MyThreading(threading.Thread):

def __init__(self,func,arg):

super(MyThreading,self).__init__()

self.func &＃61; func

self.arg &＃61; arg

def run(self):

self.func(self.arg)

def f1(args):

print(args)

obj &＃61; MyThreading(f1, 123)

obj.start()

1.3 线程锁

CPU执行任务时&＃xff0c;在线程之间是进行随机调度的&＃xff0c;并且每个线程可能只执行n条代码后就转而执行另外一条线程。由于在一个进程中的多个线程之间是共享资源和数据的&＃xff0c;这就容易造成资源抢夺或脏数据&＃xff0c;于是就有了锁的概念&＃xff0c;限制某一时刻只有一个线程能访问某个指定的数据。

1.3.1 未使用锁

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import threading

import time

NUM &＃61; 0

def show():

global NUM

NUM &＃43;&＃61; 1

name &＃61; t.getName()

time.sleep(1) # 注意&＃xff0c;这行语句的位置很重要&＃xff0c;必须在NUM被修改后&＃xff0c;否则观察不到脏数据的现象。

print(name, "执行完毕后&＃xff0c;NUM的值为&＃xff1a; ", NUM)

for i in range(10):

t &＃61; threading.Thread(target&＃61;show)

t.start()

print(&＃39;main thread stop&＃39;)

上述代码运行后&＃xff0c;结果如下&＃xff1a;

main thread stop

Thread-1 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-2 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-4 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-9 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-3 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-6 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-8 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-7 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-5 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

Thread-10 执行完毕后&＃xff0c;NUM的值为&＃xff1a; 10

由此可见&＃xff0c;由于线程同时访问一个数据&＃xff0c;产生了错误的结果。为了解决这个问题&＃xff0c;python在threading模块中定义了几种线程锁类&＃xff0c;分别是&＃xff1a;Lock 普通锁(不可嵌套)

RLock 普通锁(可嵌套)常用

Semaphore 信号量

event 事件

condition 条件

1.3.2 普通锁Lock和RLock

类名&＃xff1a;Lock或RLock

普通锁&＃xff0c;也叫互斥锁&＃xff0c;是独占的&＃xff0c;同一时刻只有一个线程被放行。

import time

import threading

NUM &＃61; 10

def func(lock):

global NUM

lock.acquire() # 让锁开始起作用

NUM -&＃61; 1

time.sleep(1)

print(NUM)

lock.release() # 释放锁

lock &＃61; threading.Lock() # 实例化一个锁对象

for i in range(10):

t &＃61; threading.Thread(target&＃61;func, args&＃61;(lock,)) # 记得把锁当作参数传递给func参数

t.start()

以上是threading模块的Lock类&＃xff0c;它不支持嵌套锁。RLcok类的用法和Lock一模一样&＃xff0c;但它支持嵌套&＃xff0c;因此我们一般直接使用RLcok类。

1.3.3 信号量(Semaphore)

类名&＃xff1a;BoundedSemaphore

这种锁允许一定数量的线程同时更改数据&＃xff0c;它不是互斥锁。比如地铁安检&＃xff0c;排队人很多&＃xff0c;工作人员只允许一定数量的人进入安检区&＃xff0c;其它的人继续排队。

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import time

import threading

def run(n):

semaphore.acquire()

print("run the thread: %s" % n)

time.sleep(1)

semaphore.release()

num &＃61; 0

semaphore &＃61; threading.BoundedSemaphore(5) # 最多允许5个线程同时运行

for i in range(20):

t &＃61; threading.Thread(target&＃61;run, args&＃61;(i,))

t.start()

1.3.4 事件(Event)

类名&＃xff1a;Event

事件主要提供了三个方法 set、wait、clear。

事件机制&＃xff1a;全局定义了一个“Flag”&＃xff0c;如果“Flag”的值为False&＃xff0c;那么当程序执行wait方法时就会阻塞&＃xff0c;如果“Flag”值为True&＃xff0c;那么wait方法时便不再阻塞。这种锁&＃xff0c;类似交通红绿灯(默认是红灯)&＃xff0c;它属于在红灯的时候一次性阻挡所有线程&＃xff0c;在绿灯的时候&＃xff0c;一次性放行所有的排队中的线程。

clear&＃xff1a;将“Flag”设置为False

set&＃xff1a;将“Flag”设置为True

import threading

def func(e,i):

print(i)

e.wait() # 检测当前event是什么状态&＃xff0c;如果是红灯&＃xff0c;则阻塞&＃xff0c;如果是绿灯则继续往下执行。默认是红灯。

print(i&＃43;100)

event &＃61; threading.Event()

for i in range(10):

t &＃61; threading.Thread(target&＃61;func, args&＃61;(event, i))

t.start()

event.clear() # 主动将状态设置为红灯

inp &＃61; input(">>>")

if inp &＃61;&＃61; "1":

event.set() # 主动将状态设置为绿灯

1.3.5 条件(condition)

类名&＃xff1a;Condition

该机制会使得线程等待&＃xff0c;只有满足某条件时&＃xff0c;才释放n个线程。

import threading

def condition():

ret &＃61; False

r &＃61; input(">>>")

if r &＃61;&＃61; "yes":

ret &＃61; True

return ret

def func(conn, i):

print(i)

conn.acquire()

conn.wait_for(condition) # 这个方法接受一个函数的返回值

print(i&＃43;100)

conn.release()

c &＃61; threading.Condition()

for i in range(10):

t &＃61; threading.Thread(target&＃61;func, args&＃61;(c, i,))

t.start()

上面的例子&＃xff0c;每输入一次“yes”放行了一个线程。下面这个&＃xff0c;可以选择一次放行几个线程。

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import threading

def run(n):

con.acquire()

con.wait()

print("run the thread: %s" %n)

con.release()

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

con &＃61; threading.Condition()

for i in range(10):

t &＃61; threading.Thread(target&＃61;run, args&＃61;(i,))

t.start()

while True:

inp &＃61; input(&＃39;>>>&＃39;)

if inp &＃61;&＃61; "q":

break

# 下面这三行是固定语法

con.acquire()

con.notify(int(inp)) # 这个方法接收一个整数&＃xff0c;表示让多少个线程通过

con.release()

1.3 全局解释器锁(GIL)

既然介绍了多线程和线程锁&＃xff0c;那就不得不提及python的GIL&＃xff0c;也就是全局解释器锁。在编程语言的世界&＃xff0c;python因为GIL的问题广受诟病&＃xff0c;因为它在解释器的层面限制了程序在同一时间只有一个线程被CPU实际执行&＃xff0c;而不管你的程序里实际开了多少条线程。所以我们经常能发现&＃xff0c;python中的多线程编程有时候效率还不如单线程&＃xff0c;就是因为这个原因。那么&＃xff0c;对于这个GIL&＃xff0c;一些普遍的问题如下&＃xff1a;

每种编程语言都有GIL吗&＃xff1f;

以python官方Cpython解释器为代表....其他语言好像未见。

为什么要有GIL&＃xff1f;

作为解释型语言&＃xff0c;Python的解释器必须做到既安全又高效。我们都知道多线程编程会遇到的问题。解释器要留意的是避免在不同的线程操作内部共享的数据。同时它还要保证在管理用户线程时总是有最大化的计算资源。那么&＃xff0c;不同线程同时访问时&＃xff0c;数据的保护机制是怎样的呢&＃xff1f;答案是解释器全局锁GIL。GIL对诸如当前线程状态和为垃圾回收而用的堆分配对象这样的东西的访问提供着保护。

为什么不能去掉GIL&＃xff1f;

首先&＃xff0c;在早期的python解释器依赖较多的全局状态&＃xff0c;传承下来&＃xff0c;使得想要移除当今的GIL变得更加困难。其次&＃xff0c;对于程序员而言&＃xff0c;仅仅是想要理解它的实现就需要对操作系统设计、多线程编程、C语言、解释器设计和CPython解释器的实现有着非常彻底的理解。

在1999年&＃xff0c;针对Python1.5&＃xff0c;一个“freethreading”补丁已经尝试移除GIL&＃xff0c;用细粒度的锁来代替。然而&＃xff0c;GIL的移除给单线程程序的执行速度带来了一定的负面影响。当用单线程执行时&＃xff0c;速度大约降低了40%。虽然使用两个线程时在速度上得到了提高&＃xff0c;但这个提高并没有随着核数的增加而线性增长。因此这个补丁没有被采纳。

另外&＃xff0c;在python的不同解释器实现中&＃xff0c;如PyPy就移除了GIL&＃xff0c;其执行速度更快(不单单是去除GIL的原因)。然而&＃xff0c;我们通常使用的CPython占有着统治地位的使用量&＃xff0c;所以&＃xff0c;你懂的。

在Python 3.2中实现了一个新的GIL&＃xff0c;并且带着一些积极的结果。这是自1992年以来&＃xff0c;GIL的一次最主要改变。旧的GIL通过对Python指令进行计数来确定何时放弃GIL。在新的GIL实现中&＃xff0c;用一个固定的超时时间来指示当前的线程以放弃这个锁。在当前线程保持这个锁&＃xff0c;且当第二个线程请求这个锁的时候&＃xff0c;当前线程就会在5ms后被强制释放掉这个锁(这就是说&＃xff0c;当前线程每5ms就要检查其是否需要释放这个锁)。当任务是可行的时候&＃xff0c;这会使得线程间的切换更加可预测。

GIL对我们有什么影响&＃xff1f;

最大的影响是我们不能随意使用多线程。要区分任务场景。

在单核cpu情况下对性能的影响可以忽略不计&＃xff0c;多线程多进程都差不多。在多核CPU时&＃xff0c;多线程效率较低。GIL对单进程和多进程没有影响。

在实际使用中有什么好的建议&＃xff1f;

建议在IO密集型任务中使用多线程&＃xff0c;在计算密集型任务中使用多进程。深入研究python的协程机制&＃xff0c;你会有惊喜的。

更多的详细介绍和说明请参考下面的文献&＃xff1a;

原文&＃xff1a;Python&＃39;s Hardest Problem

译文&＃xff1a;Python 最难的问题

1.4 定时器(Timer)

定时器&＃xff0c;指定n秒后执行某操作。很简单但很使用的东西。

from threading import Timer

def hello():

print("hello, world")

t &＃61; Timer(1, hello) # 表示1秒后执行hello函数

t.start()

1.5 队列

通常而言&＃xff0c;队列是一种先进先出的数据结构&＃xff0c;与之对应的是堆栈这种后进先出的结构。但是在python中&＃xff0c;它内置了一个queue模块&＃xff0c;它不但提供普通的队列&＃xff0c;还提供一些特殊的队列。具体如下&＃xff1a;queue.Queue &＃xff1a;先进先出队列

queue.LifoQueue &＃xff1a;后进先出队列

queue.PriorityQueue &＃xff1a;优先级队列

queue.deque &＃xff1a;双向队列

1.5.1 Queue&＃xff1a;先进先出队列

这是最常用也是最普遍的队列&＃xff0c;先看一个例子。

import queue

q &＃61; queue.Queue(5)

q.put(11)

q.put(22)

q.put(33)

print(q.get())

print(q.get())

print(q.get())

Queue类的参数和方法&＃xff1a;

maxsize 队列的最大元素个数&＃xff0c;也就是queue.Queue(5)中的5。当队列内的元素达到这个值时&＃xff0c;后来的元素默认会阻塞&＃xff0c;等待队列腾出位置。

def __init__(self, maxsize&＃61;0):

self.maxsize &＃61; maxsize

self._init(maxsize)

qsize()获取当前队列中元素的个数&＃xff0c;也就是队列的大小

empty()判断当前队列是否为空&＃xff0c;返回True或者False

full()判断当前队列是否已满&＃xff0c;返回True或者False

put(self, block&＃61;True, timeout&＃61;None)

往队列里放一个元素&＃xff0c;默认是阻塞和无时间限制的。如果&＃xff0c;block设置为False&＃xff0c;则不阻塞&＃xff0c;这时&＃xff0c;如果队列是满的&＃xff0c;放不进去&＃xff0c;就会弹出异常。如果timeout设置为n秒&＃xff0c;则会等待这个秒数后才put&＃xff0c;如果put不进去则弹出异常。

get(self, block&＃61;True, timeout&＃61;None)

从队列里获取一个元素。参数和put是一样的意思。

join()阻塞进程&＃xff0c;直到所有任务完成&＃xff0c;需要配合另一个方法task_done。

def join(self):

with self.all_tasks_done:

while self.unfinished_tasks:

self.all_tasks_done.wait()

task_done() 表示某个任务完成。每一条get语句后需要一条task_done。

import queue

q &＃61; queue.Queue(5)

q.put(11)

q.put(22)

print(q.get())

q.task_done()

print(q.get())

q.task_done()

q.join()

1.5.2 LifoQueue&＃xff1a;后进先出队列

类似于“堆栈”&＃xff0c;后进先出。也较常用。

import queue

q &＃61; queue.LifoQueue()

q.put(123)

q.put(456)

print(q.get())

上述代码运行结果是&＃xff1a;456

1.5.3 PriorityQueue&＃xff1a;优先级队列

带有权重的队列&＃xff0c;每个元素都是一个元组&＃xff0c;前面的数字表示它的优先级&＃xff0c;数字越小优先级越高&＃xff0c;同样的优先级先进先出

q &＃61; queue.PriorityQueue()

q.put((1,"alex1"))

q.put((1,"alex2"))

q.put((1,"alex3"))

q.put((3,"alex3"))

print(q.get())

1.5.4 deque&＃xff1a;双向队列

Queue和LifoQueue的“综合体”&＃xff0c;双向进出。方法较多&＃xff0c;使用复杂&＃xff0c;慎用&＃xff01;

q &＃61; queue.deque()

q.append(123)

q.append(333)

q.appendleft(456)

q.pop()

q.popleft()

1.6 生产者消费者模型

利用多线程和队列可以搭建一个生产者消费者模型&＃xff0c;用于处理大并发的服务。

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。该模式通过平衡生产线程和消费线程的工作能力来提高程序的整体处理数据的速度。

为什么要使用生产者和消费者模式

在线程世界里&＃xff0c;生产者就是生产数据的线程&＃xff0c;消费者就是消费数据的线程。在多线程开发当中&＃xff0c;如果生产者处理速度很快&＃xff0c;而消费者处理速度很慢&＃xff0c;那么生产者就必须等待消费者处理完&＃xff0c;才能继续生产数据。同样的道理&＃xff0c;如果消费者的处理能力大于生产者&＃xff0c;那么消费者就必须等待生产者。为了解决这个问题于是引入了生产者和消费者模式。

什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯&＃xff0c;而通过阻塞队列来进行通讯&＃xff0c;所以生产者生产完数据之后不用等待消费者处理&＃xff0c;直接扔给阻塞队列&＃xff0c;消费者不找生产者要数据&＃xff0c;而是直接从阻塞队列里取&＃xff0c;阻塞队列就相当于一个缓冲区&＃xff0c;平衡了生产者和消费者的处理能力。

这个阻塞队列就是用来给生产者和消费者解耦的。纵观大多数设计模式&＃xff0c;都会找一个第三者出来进行解耦&＃xff0c;如工厂模式的第三者是工厂类&＃xff0c;模板模式的第三者是模板类。在学习一些设计模式的过程中&＃xff0c;如果先找到这个模式的第三者&＃xff0c;能帮助我们快速熟悉一个设计模式。

以上摘自方腾飞的《聊聊并发——生产者消费者模式》

下面是一个简单的厨师做包子&＃xff0c;顾客吃包子的例子。

#!/usr/bin/env python

# -*- coding:utf-8 -*-

# Author:Liu Jiang

import time

import queue

import threading

q &＃61; queue.Queue(10)

def productor(i):

while True:

q.put("厨师 %s 做的包子&＃xff01;"%i)

time.sleep(2)

def consumer(k):

while True:

print("顾客 %s 吃了一个 %s"%(k,q.get()))

time.sleep(1)

for i in range(3):

t &＃61; threading.Thread(target&＃61;productor,args&＃61;(i,))

t.start()

for k in range(10):

v &＃61; threading.Thread(target&＃61;consumer,args&＃61;(k,))

v.start()

1.7 线程池

在使用多线程处理任务时也不是线程越多越好&＃xff0c;由于在切换线程的时候&＃xff0c;需要切换上下文环境&＃xff0c;依然会造成cpu的大量开销。为解决这个问题&＃xff0c;线程池的概念被提出来了。预先创建好一个较为优化的数量的线程&＃xff0c;让过来的任务立刻能够使用&＃xff0c;就形成了线程池。在python中&＃xff0c;没有内置的较好的线程池模块&＃xff0c;需要自己实现或使用第三方模块。下面是一个简单的线程池&＃xff1a;

#!/usr/bin/env python

# -*- coding:utf-8 -*-

# Author:Liu Jiang

import queue

import time

import threading

class MyThreadPool:

def __init__(self, maxsize&＃61;5):

self.maxsize &＃61; maxsize

self._q &＃61; 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)

def task(i, pool):

print(i)

time.sleep(1)

pool.add_thread()

pool &＃61; MyThreadPool(5)

for i in range(100):

t &＃61; pool.get_thread()

obj &＃61; t(target&＃61;task, args&＃61;(i,pool))

obj.start()

上面的例子是把线程类当做元素添加到队列内。实现方法比较糙&＃xff0c;每个线程使用后就被抛弃&＃xff0c;一开始就将线程开到满&＃xff0c;因此性能较差。下面是一个相对好一点的例子&＃xff1a;

#!/usr/bin/env python

# -*- coding:utf-8 -*-

import queue

import threading

import contextlib

import time

StopEvent &＃61; object() # 创建空对象

class ThreadPool(object):

def __init__(self, max_num, max_task_num &＃61; None):

if max_task_num:

self.q &＃61; queue.Queue(max_task_num)

else:

self.q &＃61; queue.Queue()

self.max_num &＃61; max_num

self.cancel &＃61; False

self.terminal &＃61; False

self.generate_list &＃61; []

self.free_list &＃61; []

def run(self, func, args, callback&＃61;None):

"""

线程池执行一个任务

:param func: 任务函数

:param args: 任务函数所需参数

:param callback: 任务执行失败或成功后执行的回调函数&＃xff0c;回调函数有两个参数1、任务函数执行状态&＃xff1b;2、任务函数返回值(默认为None&＃xff0c;即&＃xff1a;不执行回调函数)

:return: 如果线程池已经终止&＃xff0c;则返回True否则None

"""

if self.cancel:

return

if len(self.free_list) &＃61;&＃61; 0 and len(self.generate_list)

self.generate_thread()

w &＃61; (func, args, callback,)

self.q.put(w)

def generate_thread(self):

"""

创建一个线程

"""

t &＃61; threading.Thread(target&＃61;self.call)

t.start()

def call(self):

"""

循环去获取任务函数并执行任务函数

"""

current_thread &＃61; threading.currentThread

self.generate_list.append(current_thread)

event &＃61; self.q.get()

while event !&＃61; StopEvent:

func, arguments, callback &＃61; event

try:

result &＃61; func(*arguments)

success &＃61; True

except Exception as e:

success &＃61; False

result &＃61; 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 &＃61; StopEvent

else:

event &＃61; self.q.get()

else:

self.generate_list.remove(current_thread)

def close(self):

"""

执行完所有的任务后&＃xff0c;所有线程停止

"""

self.cancel &＃61; True

full_size &＃61; len(self.generate_list)

while full_size:

self.q.put(StopEvent)

full_size -&＃61; 1

def terminate(self):

"""

无论是否还有任务&＃xff0c;终止线程

"""

self.terminal &＃61; True

while self.generate_list:

self.q.put(StopEvent)

self.q.empty()

&＃64;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 &＃61; 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 &＃61; 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()

二、进程

在python中multiprocess模块提供了Process类&＃xff0c;实现进程相关的功能。但是&＃xff0c;由于它是基于fork机制的&＃xff0c;因此不被windows平台支持。想要在windows中运行&＃xff0c;必须使用if __name__ &＃61;&＃61; &＃39;__main__:的方式&＃xff0c;显然这只能用于调试和学习&＃xff0c;不能用于实际环境。

(PS&＃xff1a;在这里我必须吐槽一下python的包、模块和类的组织结构。在multiprocess中你既可以import大写的Process&＃xff0c;也可以import小写的process&＃xff0c;这两者是完全不同的东西。这种情况在python中很多&＃xff0c;新手容易傻傻分不清。)

下面是一个简单的多进程例子&＃xff0c;你会发现Process的用法和Thread的用法几乎一模一样。

from multiprocessing import Process

def foo(i):

print("This is Process ", i)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

for i in range(5):

p &＃61; Process(target&＃61;foo, args&＃61;(i,))

p.start()

2.1 进程的数据共享

每个进程都有自己独立的数据空间&＃xff0c;不同进程之间通常是不能共享数据&＃xff0c;创建一个进程需要非常大的开销。

from multiprocessing import Process

list_1 &＃61; []

def foo(i):

list_1.append(i)

print("This is Process ", i," and list_1 is ", list_1)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

for i in range(5):

p &＃61; Process(target&＃61;foo, args&＃61;(i,))

p.start()

print("The end of list_1:", list_1)

运行上面的代码&＃xff0c;你会发现列表list_1在各个进程中只有自己的数据&＃xff0c;完全无法共享。想要进程之间进行资源共享可以使用queues/Array/Manager这三个multiprocess模块提供的类。

2.1.1 使用Array共享数据

from multiprocessing import Process

from multiprocessing import Array

def Foo(i,temp):

temp[0] &＃43;&＃61; 100

for item in temp:

print(i,&＃39;----->&＃39;,item)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

temp &＃61; Array(&＃39;i&＃39;, [11, 22, 33, 44])

for i in range(2):

p &＃61; Process(target&＃61;Foo, args&＃61;(i,temp))

p.start()

对于Array数组类&＃xff0c;括号内的“i”表示它内部的元素全部是int类型&＃xff0c;而不是指字符i&＃xff0c;列表内的元素可以预先指定&＃xff0c;也可以指定列表长度。概括的来说就是Array类在实例化的时候就必须指定数组的数据类型和数组的大小&＃xff0c;类似temp &＃61; Array(&＃39;i&＃39;, 5)。对于数据类型有下面的表格对应&＃xff1a;

&＃39;c&＃39;: ctypes.c_char, &＃39;u&＃39;: ctypes.c_wchar,

&＃39;b&＃39;: ctypes.c_byte, &＃39;B&＃39;: ctypes.c_ubyte,

&＃39;h&＃39;: ctypes.c_short, &＃39;H&＃39;: ctypes.c_ushort,

&＃39;i&＃39;: ctypes.c_int, &＃39;I&＃39;: ctypes.c_uint,

&＃39;l&＃39;: ctypes.c_long, &＃39;L&＃39;: ctypes.c_ulong,

&＃39;f&＃39;: ctypes.c_float, &＃39;d&＃39;: ctypes.c_double

2.1.2 使用Manager共享数据

from multiprocessing import Process,Manager

def Foo(i,dic):

dic[i] &＃61; 100&＃43;i

print(dic.values())

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

manage &＃61; Manager()

dic &＃61; manage.dict()

for i in range(10):

p &＃61; Process(target&＃61;Foo, args&＃61;(i,dic))

p.start()

p.join()

Manager比Array要好用一点&＃xff0c;因为它可以同时保存多种类型的数据格式。

2.1.3 使用queues的Queue类共享数据

import multiprocessing

from multiprocessing import Process

from multiprocessing import queues

def foo(i,arg):

arg.put(i)

print(&＃39;The Process is &＃39;, i, "and the queue&＃39;s size is ", arg.qsize())

if __name__ &＃61;&＃61; "__main__":

li &＃61; queues.Queue(20, ctx&＃61;multiprocessing)

for i in range(10):

p &＃61; Process(target&＃61;foo, args&＃61;(i,li,))

p.start()

这里就有点类似上面的队列了。从运行结果里&＃xff0c;你还能发现数据共享中存在的脏数据问题。另外&＃xff0c;比较悲催的是multiprocessing里还有一个Queue&＃xff0c;一样能实现这个功能。

2.2 进程锁

为了防止和多线程一样的出现数据抢夺和脏数据的问题&＃xff0c;同样需要设置进程锁。与threading类似&＃xff0c;在multiprocessing里也有同名的锁类RLock, Lock, Event, Condition, Semaphore&＃xff0c;连用法都是一样样的&＃xff01;(这个我喜欢)

from multiprocessing import Process

from multiprocessing import queues

from multiprocessing import Array

from multiprocessing import RLock, Lock, Event, Condition, Semaphore

import multiprocessing

import time

def foo(i,lis,lc):

lc.acquire()

lis[0] &＃61; lis[0] - 1

time.sleep(1)

print(&＃39;say hi&＃39;,lis[0])

lc.release()

if __name__ &＃61;&＃61; "__main__":

# li &＃61; []

li &＃61; Array(&＃39;i&＃39;, 1)

li[0] &＃61; 10

lock &＃61; RLock()

for i in range(10):

p &＃61; Process(target&＃61;foo,args&＃61;(i,li,lock))

p.start()

2.3 进程池

既然有线程池&＃xff0c;那必然也有进程池。但是&＃xff0c;python给我们内置了一个进程池&＃xff0c;不需要像线程池那样需要自定义&＃xff0c;你只需要简单的from multiprocessing import Pool。

#!/usr/bin/env python

# -*- coding:utf-8 -*-

from multiprocessing import Pool

import time

def f1(args):

time.sleep(1)

print(args)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

p &＃61; Pool(5)

for i in range(30):

p.apply_async(func&＃61;f1, args&＃61; (i,))

p.close() # 等子进程执行完毕后关闭线程池

# time.sleep(2)

# p.terminate() # 立刻关闭线程池

p.join()

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

进程池中有以下几个主要方法&＃xff1a;apply&＃xff1a;从进程池里取一个进程并执行

apply_async&＃xff1a;apply的异步版本

terminate:立刻关闭线程池

join&＃xff1a;主进程等待所有子进程执行完毕&＃xff0c;必须在close或terminate之后

close&＃xff1a;等待所有进程结束后&＃xff0c;才关闭线程池

三、协程

线程和进程的操作是由程序触发系统接口&＃xff0c;最后的执行者是系统&＃xff0c;它本质上是操作系统提供的功能。而协程的操作则是程序员指定的&＃xff0c;在python中通过yield&＃xff0c;人为的实现并发处理。

协程存在的意义&＃xff1a;对于多线程应用&＃xff0c;CPU通过切片的方式来切换线程间的执行&＃xff0c;线程切换时需要耗时。协程&＃xff0c;则只使用一个线程&＃xff0c;分解一个线程成为多个“微线程”&＃xff0c;在一个线程中规定某个代码块的执行顺序。

协程的适用场景&＃xff1a;当程序中存在大量不需要CPU的操作时(IO)。

在不需要自己“造轮子”的年代&＃xff0c;同样有第三方模块为我们提供了高效的协程&＃xff0c;这里介绍一下greenlet和gevent。本质上&＃xff0c;gevent是对greenlet的高级封装&＃xff0c;因此一般用它就行&＃xff0c;这是一个相当高效的模块。

在使用它们之前&＃xff0c;需要先安装&＃xff0c;可以通过源码&＃xff0c;也可以通过pip。

3.1 greenlet

from greenlet import greenlet

def test1():

print(12)

gr2.switch()

print(34)

gr2.switch()

def test2():

print(56)

gr1.switch()

print(78)

gr1 &＃61; greenlet(test1)

gr2 &＃61; greenlet(test2)

gr1.switch()

实际上&＃xff0c;greenlet就是通过switch方法在不同的任务之间进行切换。

3.2 gevent

from gevent import monkey; monkey.patch_all()

import gevent

import requests

def f(url):

print(&＃39;GET: %s&＃39; % url)

resp &＃61; requests.get(url)

data &＃61; resp.text

print(&＃39;%d bytes received from %s.&＃39; % (len(data), url))

gevent.joinall([

gevent.spawn(f, &＃39;https://www.python.org/&＃39;),

gevent.spawn(f, &＃39;https://www.yahoo.com/&＃39;),

gevent.spawn(f, &＃39;https://github.com/&＃39;),

])

通过joinall将任务f和它的参数进行统一调度&＃xff0c;实现单线程中的协程。代码封装层次很高&＃xff0c;实际使用只需要了解它的几个主要方法即可。

更多python线程、进程和协程相关文章请关注PHP中文网&＃xff01;

本文原创发布php中文网&＃xff0c;转载请注明出处&＃xff0c;感谢您的尊重&＃xff01;