python线程、进程和协程详解

引言

解释器环境:python3.5.1

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

server = socketserver.ThreadingTCPServer(settings.IP_PORT, MyServer)

ThreadingTCPServer这个类是一个支持多线程和TCP协议的socketserver,它的继承关系是这样的:
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass

右边的TCPServer实际上是它主要的功能父类,而左边的ThreadingMixIn则是实现了多线程的类,它自己本身则没有任何代码。
MixIn在python的类命名中,很常见,一般被称为“混入”,戏称“乱入”,通常为了某种重要功能被子类继承。

class ThreadingMixIn:

 daemon_threads = 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 = threading.Thread(target = self.process_request_thread,
        args = (request, client_address))
  t.daemon = self.daemon_threads
  t.start()

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

一、线程

代码如下:

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

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

1.1 普通的多线程

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

import threading
import time

def show(arg):
 time.sleep(1)
 print('thread'+str(arg))

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

print('main thread stop')

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

下面是Thread类的主要方法:

start 线程准备就绪,等待CPU调度
setName 为线程设置名称
getName 获取线程名称
setDaemon 设置为后台线程或前台线程(默认)
如果是后台线程,主线程执行过程中,后台线程也在进行,主线程执行完毕后,后台线程不论成功与否,均停止。如果是前台线程,主线程执行过程中,前台线程也在进行,主线程执行完毕后,等待前台线程也执行完成后,程序停止。
join 逐个执行每个线程,执行完毕后继续往下执行,该方法使得多线程变得无意义。
run 线程被cpu调度后自动执行线程对象的run方法  

1.2 自定义线程类

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

import threading
class MyThreading(threading.Thread):

 def __init__(self,func,arg):
  super(MyThreading,self).__init__()
  self.func = func
  self.arg = arg

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

def f1(args):
 print(args)

obj = MyThreading(f1, 123)
obj.start()

1.3 线程锁

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

1.3.1 未使用锁

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time
NUM = 0
def show():
 global NUM
 NUM += 1
 name = t.getName()
 time.sleep(1)  # 注意,这行语句的位置很重要,必须在NUM被修改后,否则观察不到脏数据的现象。
 print(name, "执行完毕后,NUM的值为: ", NUM)

for i in range(10):
 t = threading.Thread(target=show)
 t.start()
print('main thread stop')

上述代码运行后,结果如下:

main thread stop
Thread-1 执行完毕后,NUM的值为: 10
Thread-2 执行完毕后,NUM的值为: 10
Thread-4 执行完毕后,NUM的值为: 10
Thread-9 执行完毕后,NUM的值为: 10
Thread-3 执行完毕后,NUM的值为: 10
Thread-6 执行完毕后,NUM的值为: 10
Thread-8 执行完毕后,NUM的值为: 10
Thread-7 执行完毕后,NUM的值为: 10
Thread-5 执行完毕后,NUM的值为: 10
Thread-10 执行完毕后,NUM的值为: 10

由此可见,由于线程同时访问一个数据,产生了错误的结果。为了解决这个问题,python在threading模块中定义了几种线程锁类,分别是:

  1. Lock 普通锁(不可嵌套)
  2. RLock 普通锁(可嵌套)常用
  3. Semaphore 信号量
  4. event 事件
  5. condition 条件

1.3.2 普通锁Lock和RLock

类名:Lock或RLock

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

import time
import threading

NUM = 10
def func(lock):
 global NUM
 lock.acquire() # 让锁开始起作用
 NUM -= 1
 time.sleep(1)
 print(NUM)
 lock.release() # 释放锁

lock = threading.Lock() # 实例化一个锁对象
for i in range(10):
 t = threading.Thread(target=func, args=(lock,)) # 记得把锁当作参数传递给func参数
 t.start()

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

1.3.3 信号量(Semaphore)

类名:BoundedSemaphore

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

#!/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 = 0
semaphore = threading.BoundedSemaphore(5) # 最多允许5个线程同时运行
for i in range(20):
 t = threading.Thread(target=run, args=(i,))
 t.start()

1.3.4 事件(Event)

类名:Event

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

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

import threading

def func(e,i):
 print(i)
 e.wait() # 检测当前event是什么状态,如果是红灯,则阻塞,如果是绿灯则继续往下执行。默认是红灯。
 print(i+100)

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

event.clear() # 主动将状态设置为红灯
inp = input(">>>")
if inp == "1":
 event.set() # 主动将状态设置为绿灯

1.3.5 条件(condition)

类名:Condition

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

import threading

def condition():
 ret = False
 r = input(">>>")
 if r == "yes":
  ret = True
 return ret

def func(conn, i):
 print(i)
 conn.acquire()
 conn.wait_for(condition) # 这个方法接受一个函数的返回值
 print(i+100)
 conn.release()

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

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

#!/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__ == '__main__':

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

 while True:
  inp = input('>>>')
  if inp == "q":
   break
  # 下面这三行是固定语法
  con.acquire()
  con.notify(int(inp)) # 这个方法接收一个整数,表示让多少个线程通过
  con.release()

1.3 全局解释器锁(GIL)

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

每种编程语言都有GIL吗?

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

为什么要有GIL?

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

为什么不能去掉GIL?

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

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

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

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

GIL对我们有什么影响?

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

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

在实际使用中有什么好的建议?

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

更多的详细介绍和说明请参考下面的文献:

原文:Python's Hardest Problem
译文:Python 最难的问题

1.4 定时器(Timer)

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

from threading import Timer
def hello():
 print("hello, world")
t = Timer(1, hello) # 表示1秒后执行hello函数
t.start()

1.5 队列

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

  1. queue.Queue :先进先出队列
  2. queue.LifoQueue :后进先出队列
  3. queue.PriorityQueue :优先级队列
  4. queue.deque :双向队列

1.5.1 Queue:先进先出队列

这是最常用也是最普遍的队列,先看一个例子。

import queue
q = queue.Queue(5)
q.put(11)
q.put(22)
q.put(33)

print(q.get())
print(q.get())
print(q.get())

Queue类的参数和方法:

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

def __init__(self, maxsize=0):
self.maxsize = maxsize
self._init(maxsize)

qsize() 获取当前队列中元素的个数,也就是队列的大小
empty() 判断当前队列是否为空,返回True或者False
full() 判断当前队列是否已满,返回True或者False
put(self, block=True, timeout=None)

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

get(self, block=True, timeout=None)

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

join() 阻塞进程,直到所有任务完成,需要配合另一个方法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 = 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:后进先出队列

类似于“堆栈”,后进先出。也较常用。

import queue
q = queue.LifoQueue()
q.put(123)
q.put(456)
print(q.get())

上述代码运行结果是:456

1.5.3 PriorityQueue:优先级队列

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

q = 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:双向队列

Queue和LifoQueue的“综合体”,双向进出。方法较多,使用复杂,慎用!

q = queue.deque()
q.append(123)
q.append(333)
q.appendleft(456)
q.pop()
q.popleft()

1.6 生产者消费者模型

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

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

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

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

什么是生产者消费者模式

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

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

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

下面是一个简单的厨师做包子,顾客吃包子的例子。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang
import time
import queue
import threading

q = queue.Queue(10)
def productor(i):
 while True:
  q.put("厨师 %s 做的包子!"%i)
  time.sleep(2)

def consumer(k):
 while True:
  print("顾客 %s 吃了一个 %s"%(k,q.get()))
  time.sleep(1)

for i in range(3):
 t = threading.Thread(target=productor,args=(i,))
 t.start()

for k in range(10):
 v = threading.Thread(target=consumer,args=(k,))
 v.start()

1.7 线程池

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

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang

import queue
import time
import threading

class MyThreadPool:
 def __init__(self, maxsize=5):
  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)

def task(i, pool):
 print(i)
 time.sleep(1)
 pool.add_thread()

pool = MyThreadPool(5)
for i in range(100):
 t = pool.get_thread()
 obj = t(target=task, args=(i,pool))
 obj.start()

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

#!/usr/bin/env python
# -*- coding:utf-8 -*-

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()

二、进程

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

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

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

from multiprocessing import Process

def foo(i):
 print("This is Process ", i)

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

2.1 进程的数据共享

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

from multiprocessing import Process
list_1 = []
def foo(i):
 list_1.append(i)
 print("This is Process ", i," and list_1 is ", list_1)

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

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

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

2.1.1 使用Array共享数据

from multiprocessing import Process
from multiprocessing import Array

def Foo(i,temp):
 temp[0] += 100
 for item in temp:
  print(i,'----->',item)

if __name__ == '__main__':
 temp = Array('i', [11, 22, 33, 44])
 for i in range(2):
  p = Process(target=Foo, args=(i,temp))
  p.start()

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

'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

2.1.2 使用Manager共享数据

from multiprocessing import Process,Manager

def Foo(i,dic):
 dic[i] = 100+i
 print(dic.values())

if __name__ == '__main__':
 manage = Manager()
 dic = manage.dict()
 for i in range(10):
  p = Process(target=Foo, args=(i,dic))
  p.start()
  p.join()

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

2.1.3 使用queues的Queue类共享数据

import multiprocessing
from multiprocessing import Process
from multiprocessing import queues

def foo(i,arg):
 arg.put(i)
 print('The Process is ', i, "and the queue's size is ", 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()

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

2.2 进程锁

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

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] = lis[0] - 1
 time.sleep(1)
 print('say hi',lis[0])
 lc.release()

if __name__ == "__main__":
 # li = []
 li = Array('i', 1)
 li[0] = 10
 lock = RLock()
 for i in range(10):
  p = Process(target=foo,args=(i,li,lock))
  p.start()

2.3 进程池

既然有线程池,那必然也有进程池。但是,python给我们内置了一个进程池,不需要像线程池那样需要自定义,你只需要简单的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__ == '__main__':
 p = Pool(5)
 for i in range(30):
  p.apply_async(func=f1, args= (i,))
 p.close()   # 等子进程执行完毕后关闭线程池
 # time.sleep(2)
 # p.terminate()  # 立刻关闭线程池
 p.join()

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

进程池中有以下几个主要方法:

  1. apply:从进程池里取一个进程并执行
  2. apply_async:apply的异步版本
  3. terminate:立刻关闭线程池
  4. join:主进程等待所有子进程执行完毕,必须在close或terminate之后
  5. close:等待所有进程结束后,才关闭线程池

三、协程

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

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

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

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

在使用它们之前,需要先安装,可以通过源码,也可以通过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 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

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

3.2 gevent

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/'),
])

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

(0)

相关推荐

  • python并发编程之多进程、多线程、异步和协程详解

    最近学习python并发,于是对多进程.多线程.异步和协程做了个总结. 一.多线程 多线程就是允许一个进程内存在多个控制权,以便让多个函数同时处于激活状态,从而让多个函数的操作同时运行.即使是单CPU的计算机,也可以通过不停地在不同线程的指令间切换,从而造成多线程同时运行的效果. 多线程相当于一个并发(concunrrency)系统.并发系统一般同时执行多个任务.如果多个任务可以共享资源,特别是同时写入某个变量的时候,就需要解决同步的问题,比如多线程火车售票系统:两个指令,一个指令检查票是否卖完

  • Tornado协程在python2.7如何返回值(实现方法)

    错误写法 class RemoteHandler(web.RequestHandler): @gen.coroutine def get(self): response = httpclient('http://www.baidu.com') self.write(response.body) @gen.coroutine def httpClient(url): result = yield httpclient.AsyncHTTPClient().fetch(url) return resu

  • Python协程的用法和例子详解

    从句法上看,协程与生成器类似,都是定义体中包含 yield 关键字的函数.可是,在协程中, yield 通常出现在表达式的右边(例如, datum = yield),可以产出值,也可以不产出 -- 如果 yield 关键字后面没有表达式,那么生成器产出 None. 协程可能会从调用方接收数据,不过调用方把数据提供给协程使用的是 .send(datum) 方法,而不是next(-) 函数. ==yield 关键字甚至还可以不接收或传出数据.不管数据如何流动, yield 都是一种流程控制工具,使用

  • 简述Python中的进程、线程、协程

    进程.线程和协程之间的关系和区别也困扰我一阵子了,最近有一些心得,写一下. 进程拥有自己独立的堆和栈,既不共享堆,亦不共享栈,进程由操作系统调度. 线程拥有自己独立的栈和共享的堆,共享堆,不共享栈,线程亦由操作系统调度(标准线程是的). 协程和线程一样共享堆,不共享栈,协程由程序员在协程的代码里显示调度. 进程和其他两个的区别还是很明显的. 协程和线程的区别是:协程避免了无意义的调度,由此可以提高性能,但也因此,程序员必须自己承担调度的责任,同时,协程也失去了标准线程使用多CPU的能力. Pyt

  • python协程用法实例分析

    本文实例讲述了python协程用法.分享给大家供大家参考.具体如下: 把函数编写为一个任务,从而能处理发送给他的一系列输入,这种函数称为协程 def print_matchs(matchtext): print "looking for",matchtext while True: line = (yield) #用 yield语句并以表达式(yield)的形式创建协程 if matchtext in line: print line >>> matcher = pr

  • python简单线程和协程学习心得(分享)

    python中对线程的支持的确不够,不过据说python有足够完备的异步网络框架模块,希望日后能学习到,这里就简单的对python中的线程做个总结 threading库可用来在单独的线程中执行任意的python可调用对象.尽管此模块对线程相关操作的支持不够,但是我们还是能够用简单的线程来处理I/O操作,以减低程序响应时间. from threading import Thread import time def countdown(n): while n > 0: print('T-minus:

  • 简单介绍Python的Tornado框架中的协程异步实现原理

    Tornado 4.0 已经发布了很长一段时间了, 新版本广泛的应用了协程(Future)特性. 我们目前已经将 Tornado 升级到最新版本, 而且也大量的使用协程特性. 很长时间没有更新博客, 今天就简单介绍下 Tornado 协程实现原理, Tornado 的协程是基于 Python 的生成器实现的, 所以首先来回顾下生成器. 生成器 Python 的生成器可以保存执行状态 并在下次调用的时候恢复, 通过在函数体内使用 yield 关键字 来创建一个生成器, 通过内置函数 next 或生

  • python 生成器协程运算实例

    一.yield运行方式 我们定义一个如下的生成器: def put_on(name): print("Hi {}, 货物来了,准备搬到仓库!".format(name)) while True: goods = yield print("货物[%s]已经被%s搬进仓库了."%(goods,name)) p = put_on("bigberg") #输出 G:\python\install\python.exe G:/python/untitled

  • 深入浅析python中的多进程、多线程、协程

    进程与线程的历史 我们都知道计算机是由硬件和软件组成的.硬件中的CPU是计算机的核心,它承担计算机的所有任务. 操作系统是运行在硬件之上的软件,是计算机的管理者,它负责资源的管理和分配.任务的调度. 程序是运行在系统上的具有某种功能的软件,比如说浏览器,音乐播放器等. 每次执行程序的时候,都会完成一定的功能,比如说浏览器帮我们打开网页,为了保证其独立性,就需要一个专门的管理和控制执行程序的数据结构--进程控制块. 进程就是一个程序在一个数据集上的一次动态执行过程. 进程一般由程序.数据集.进程控

  • python线程、进程和协程详解

    引言 解释器环境:python3.5.1 我们都知道python网络编程的两大必学模块socket和socketserver,其中的socketserver是一个支持IO多路复用和多线程.多进程的模块.一般我们在socketserver服务端代码中都会写这么一句: server = socketserver.ThreadingTCPServer(settings.IP_PORT, MyServer) ThreadingTCPServer这个类是一个支持多线程和TCP协议的socketserver

  • Python全栈之协程详解

    目录 1. 线程队列 2. 进程池_线程池 3. 回调函数 4. 协程 总结: 1. 线程队列 # ### 线程队列 from queue import Queue """ put 存放 超出队列长度阻塞 get 获取 超出队列长度阻塞 put_nowait 存放,超出队列长度报错 get_nowait 获取,超出队列长度报错 """ # (1) Queue """先进先出,后进先出"""

  • Python 高级教程之线程进程和协程的代码解析

    目录 进程 进程 5 种基本状态 进程的特点 进程间数据共享 进程池 进程的缺点 线程 线程的定义 使用线程模块的简单示例 代码解析 协程 协程与线程 Python 协程 协程的执行 关闭协程 链接协程以创建管道 总结 进程 进程是指在系统中正在运行的一个应用程序,是 CPU 的最小工作单元. 进程 5 种基本状态 一个进程至少具有 5 种基本状态:初始态.就绪状态.等待(阻塞)状态.执行状态.终止状态. 初始状态:进程刚被创建,由于其他进程正占有CPU资源,所以得不到执行,只能处于初始状态.

  • C++ Boost Coroutine使用协程详解

    目录 一.说明语言扩展 二.库Boost.Coroutine 三.示例和代码 一.说明语言扩展 以下库扩展了编程语言 C++. Boost.Coroutine 使得在 C++ 中使用协程成为可能——其他编程语言通常通过关键字 yield 支持. Boost.Foreach 提供了一个基于范围的 for 循环,它是在 C++11 中添加到语言中的. Boost.Parameter 允许您以名称/值对的形式并以任何顺序传递参数——例如,这在 Python 中是允许的. Boost.Conversio

  • Python3.10 Generator生成器Coroutine原生协程详解

    目录 引言 协程底层实现 业务场景 结语 引言 普遍意义上讲,生成器是一种特殊的迭代器,它可以在执行过程中暂停并在恢复执行时保留它的状态.而协程,则可以让一个函数在执行过程中暂停并在恢复执行时保留它的状态,在Python3.10中,原生协程的实现手段,就是生成器,或者说的更具体一些:协程就是一种特殊的生成器,而生成器,就是协程的入门心法. 协程底层实现 我们知道,Python3.10中可以使用async和await关键字来实现原生协程函数的定义和调度,但其实,我们也可以利用生成器达到协程的效果,

  • Python进阶之协程详解

    目录 协程 协程的应用场景 抢占式调度的缺点 用户态协同调度的优势 协程的运行原理 Python 中的协程 总结 协程 协程(co-routine,又称微线程)是一种多方协同的工作方式.当前执行者在某个时刻主动让出(yield)控制流,并记住自身当前的状态,以便在控制流返回时能从上次让出的位置恢复(resume)执行. 简而言之,协程的核心思想就在于执行者对控制流的 “主动让出” 和 “恢复”.相对于,线程此类的 “抢占式调度” 而言,协程是一种 “协作式调度” 方式. 协程的应用场景 抢占式调

  • go语言中的协程详解

    协程的特点 1.该任务的业务代码主动要求切换,即主动让出执行权限 2.发生了IO,导致执行阻塞(使用channel让协程阻塞) 与线程本质的不同 C#.java中我们执行多个线程,是通过时间片切换来进行的,要知道进行切换,程序需要保存上下文等信息,是比较消耗性能的 GO语言中的协程,没有上面这种切换,一定是通过协程主动放出权限,不是被动的. 例如: C# 中创建两个线程 可以看到1和2是交替执行的 Go语言中用协程实现一下 runtime.GOMAXPROCS(1) 这个结果就是 执行了1 在执

  • python 线程的暂停, 恢复, 退出详解及实例

    python 线程 暂停, 恢复, 退出 我们都知道python中可以是threading模块实现多线程, 但是模块并没有提供暂停, 恢复和停止线程的方法, 一旦线程对象调用start方法后, 只能等到对应的方法函数运行完毕. 也就是说一旦start后, 线程就属于失控状态. 不过, 我们可以自己实现这些. 一般的方法就是循环地判断一个标志位, 一旦标志位到达到预定的值, 就退出循环. 这样就能做到退出线程了. 但暂停和恢复线程就有点难了, 我一直也不清除有什么好的方法, 直到我看到thread

  • Python线程障碍对象Barrier原理详解

    python线程Barrier俗称障碍对象,也称栅栏,也叫屏障. 一.线程障碍对象Barrier简介 # 导入线程模块 import threading # 障碍对象barrier barrier = threading.Barrier(parties, action=None, timeout=None) parties - 线程计数器,记录线程数量,也称线程障碍数量: action - 是一个可调用函数,当等待的线程到达了线程障碍数量parties,其中一个线程会首先调用action 对应函

随机推荐