Python3 系列之并行编程

布满荆棘的人生 2023-08-17 15:35 139阅读 0赞

## 进程和线程 ##

进程是程序运行的实例。一个进程里面可以包含多个线程，因此同一进程下的多个线程之间可以共享线程内的所有资源，它是操作系统动态运行的基本单元；每一个线程是进程下的一个实例，可以动态调度和独立运行，由于线程和进程有很多类似的特点，因此，线程又被称为轻量级的进程。线程的运行在进程之下，进程的存在依赖于线程；

## 开胃菜 ##

基于 **Python3** 创建一个简单的进程示例

from threading import Thread from time import sleep class CookBook(Thread): def __init__(self): Thread.__init__(self) self.message = "Hello Parallel Python CookBook!!\n" def print_message(self): print(self.message) def run(self): print("Thread Starting\n") x = 0 while x < 10: self.print_message() sleep(2) x += 1 print("Thread Ended!\n") print("Process Started") hello_python = CookBook() hello_python.start() print("Process Ended")

> 需要注意的是，永远不要让线程在后台默默执行，当其执行完毕后要及时释放资源。

## 基于线程的并行 ##

多线程编程一般使用共享内存空间进行线程间的通信，这就使管理内存空间成为多线程编程的关键。Python 通过标准库 **threading** 模块来管理线程，具有以下的组件：

*  线程对象
 *  Lock 对象
 *  RLock 对象
 *  信号对象
 *  条件对象
 *  事件对象

## 定义一个线程 ##

### 基本语法 ###

示例代码如下所示

import threading def function(i): print("function called by thread: {0}".format(i)) return threads = [] for i in range(5): t = threading.Thread(target=function, args=(i,)) threads.append(t) t.start() lambda t, threads: t.join()

> 需要注意的是，线程创建后并不会自动运行，需要主动调用 **start()** 方法来启动线程，**join()** 会让调用它的线程被阻塞直到执行结束。（PS：可通过调用 **t.setDaemon(True)** 使其为后台线程避免主线程被阻塞）

### 线程定位 ###

示例代码如下所示

import threading import time def first_function(): print("{0} is starting".format(threading.currentThread().getName())) time.sleep(2) print("{0} is Exiting".format(threading.currentThread().getName())) def second_function(): print("{0} is starting".format(threading.currentThread().getName())) time.sleep(2) print("{0} is Exiting".format(threading.currentThread().getName())) def third_function(): print("{0} is starting".format(threading.currentThread().getName())) time.sleep(2) print("{0} is Exiting".format(threading.currentThread().getName())) if __name__ == "__main__": t1 = threading.Thread(target=first_function,name="first") t2 = threading.Thread(target=second_function,name="second") t3 = threading.Thread(target=third_function,name="third") t1.start() t2.start() t3.start() t1.join() t2.join() t3.join()

> 通过设置 **threading.Thread()** 函数的 *name* 参数来设置线程名称，通过 **threading.currentThread().getName()** 来获取当前线程名称；线程的默认名称会以 **Thread-\{i\}** 格式来定义

### 自定义一个线程对象 ###

示例代码如下所示

import threading import time exitFlag = 0 class myThread(threading.Thread): def __init__(self, threadID, name, counter): threading.Thread.__init__(self) self.threadID = threadID self.name = name self.counter = counter def run(self): print("Starting:{0}".format(self.name)) print_time(self.name, self.counter, 5) print("Exiting:{0}".format(self.name)) def print_time(threadName, delay, counter): while counter: if exitFlag: thread.exit() time.sleep(delay) print("{0} {1}".format(threadName, time.ctime(time.time()))) counter -= 1 t1 = myThread(1, "Thread-1", 1) t2 = myThread(2, "Thread-2", 1) t1.start() t2.start() t1.join() t2.join() print("Exiting Main Thread.")

> 如果想自定义一个线程对象，首先就是要定义一个继承 **threading.Thread** 类的子类，实现**构造函数**， 并重写 **run()** 方法即可。

### 线程同步 ###

#### Lock ####

示例代码如下所示

import threading shared_resource_with_lock = 0 shared_resource_with_no_lock = 0 COUNT = 100000 shared_resource_lock = threading.Lock() def increment_with_lock(): global shared_resource_with_lock for i in range(COUNT): shared_resource_lock.acquire() shared_resource_with_lock += 1 shared_resource_lock.release() def decrement_with_lock(): global shared_resource_with_lock for i in range(COUNT): shared_resource_lock.acquire() shared_resource_with_lock -= 1 shared_resource_lock.release() def increment_without_lock(): global shared_resource_with_no_lock for i in range(COUNT): shared_resource_with_no_lock += 1 def decrement_wthout_lock(): global shared_resource_with_no_lock for i in range(COUNT): shared_resource_with_no_lock -= 1 if __name__ == "__main__": t1 = threading.Thread(target=increment_with_lock) t2 = threading.Thread(target=decrement_with_lock) t3 = threading.Thread(target=increment_without_lock) t4 = threading.Thread(target=decrement_wthout_lock) t1.start() t2.start() t3.start() t4.start() t1.join() t2.join() t3.join() t4.join() print("the value of shared variable with lock management is :{0}".format( shared_resource_with_lock)) print("the value of shared variable with race condition is :{0}".format( shared_resource_with_no_lock))

> 通过 **threading.Lock()** 方法我们可以拿到线程锁，一般有两种操作方式：**acquire()** 和 **release()** 在两者之间是加锁状态，如果释放失败的话会显示 **RuntimError()** 的异常。

#### RLock ####

RLock 也叫递归锁，和 **Lock** 的区别在于：谁拿到谁释放,是通过 **threading.RLock()** 来拿到的；

示例代码如下所示

import threading import time class Box(object): lock = threading.RLock() def __init__(self): self.total_items = 0 def execute(self, n): Box.lock.acquire() self.total_items += n Box.lock.release() def add(self): Box.lock.acquire() self.execute(1) Box.lock.release() def remove(self): Box.lock.acquire() self.execute(-1) Box.lock.release() def adder(box, items): while items > 0: print("adding 1 item in the box") box.add() time.sleep(1) items -= 1 def remover(box, items): while items > 0: print("removing 1 item in the box") box.remove() time.sleep(1) items -= 1 if __name__ == "__main__": items = 5 print("putting {0} items in the box".format(items)) box = Box() t1 = threading.Thread(target=adder, args=(box, items)) t2 = threading.Thread(target=remover, args=(box, items)) t1.start() t2.start() t1.join() t2.join() print("{0} items still remain in the box".format(box.total_items))

#### 信号量 ####

示例代码如下所示

import threading import time import random semaphore = threading.Semaphore(0) def consumer(): print("Consumer is waiting.") semaphore.acquire() print("Consumer notify:consumed item numbers {0}".format(item)) def producer(): global item time.sleep(10) item = random.randint(0, 10000) print("producer notify:produced item number {0}".format(item)) semaphore.release() if __name__ == "__main__": for i in range(0, 5): t1 = threading.Thread(target=producer) t2 = threading.Thread(target=consumer) t1.start() t2.start() t1.join() t2.join() print("program terminated.")

> 信号量初始化为 0 ，然后在两个并行线程中，通过调用 **semaphore.acquire()** 函数会阻塞消费者线程，直到 **semaphore.release()** 在生产者中被调用，这里模拟了**生产者-消费者** 模式来进行了测试；如果信号量的计数器到了0，就会阻塞 acquire() 方法，直到得到另一个线程的通知。如果信号量的计数器大于0，就会对这个值-1然后分配资源。

#### 使用条件进行线程同步 ####

解释条件机制最好的例子还是生产者-消费者问题。在本例中，只要缓存不满，生产者一直向缓存生产；只要缓存不空，消费者一直从缓存取出（之后销毁）。当缓冲队列不为空的时候，生产者将通知消费者；当缓冲队列不满的时候，消费者将通知生产者。

示例代码如下所示

from threading import Thread, Condition import time items = [] condition = Condition() class consumer(Thread): def __init__(self): Thread.__init__(self) def consume(self): global condition global items condition.acquire() if len(items) == 0: condition.wait() print("Consumer notify:no item to consum") items.pop() print("Consumer notify: consumed 1 item") print("Consumer notify: item to consume are:{0}".format(len(items))) condition.notify() condition.release() def run(self): for i in range(0, 20): time.sleep(2) self.consume() class producer(Thread): def __init__(self): Thread.__init__(self) def produce(self): global condition global items condition.acquire() if len(items) == 10: condition.wait() print("Producer notify:items producted are:{0}".format(len(items))) print("Producer notify:stop the production!!") items.append(1) print("Producer notify:total items producted:{0}".format(len(items))) condition.notify() condition.release() def run(self): for i in range(0, 20): time.sleep(1) self.produce() if __name__ == "__main__": producer = producer() consumer = consumer() producer.start() consumer.start() producer.join() consumer.join()

> 通过 **condition.acquire()** 来获取锁对象，**condition.wait()** 会使当前线程进入阻塞状态，直到收到 **condition.notify()** 信号，同时，调用信号的通知的对象也要及时调用 **condition.release()** 来释放资源；

#### 使用事件进行线程同步 ####

事件是线程之间用于通信的对。有的线程等待信号，有的线程发出信号。

示例代码如下所示

import time from threading import Thread, Event import random items = [] event = Event() class consumer(Thread): def __init__(self, items, event): Thread.__init__(self) self.items = items self.event = event def run(self): while True: time.sleep(2) self.event.wait() item = self.items.pop() print('Consumer notify:{0} popped from list by {1}'.format( item, self.name)) class producer(Thread): def __init__(self, integers, event): Thread.__init__(self) self.items = items self.event = event def run(self): global item for i in range(100): time.sleep(2) item = random.randint(0, 256) self.items.append(item) print('Producer notify: item N° %d appended to list by %s' % (item, self.name)) print('Producer notify: event set by %s' % self.name) self.event.set() print('Produce notify: event cleared by %s ' % self.name) self.event.clear() if __name__ == "__main__": t1 = producer(items, event) t2 = consumer(items, event) t1.start() t2.start() t1.join() t2.join()

#### 使用 **with** 语法简化代码 ####

import threading import logging logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s') def threading_with(statement): with statement: logging.debug("%s acquired via with" % statement) def Threading_not_with(statement): statement.acquire() try: logging.debug("%s acquired directly " % statement) finally: statement.release() if __name__ == "__main__": lock = threading.Lock() rlock = threading.RLock() condition = threading.Condition() mutex = threading.Semaphore(1) threading_synchronization_list = [lock, rlock, condition, mutex] for statement in threading_synchronization_list: t1 = threading.Thread(target=threading_with, args=(statement,)) t2 = threading.Thread(target=Threading_not_with, args=(statement,)) t1.start() t2.start() t1.join() t2.join()

#### 使用 **queue** 进行线程通信 ####

**Queue** 常用的方法有以下四个：

*  put()：往 queue 中添加一个元素
 *  get()：从 queue 中删除一个元素，并返回该元素
 *  task\_done()：每次元素被处理的时候都需要调用这个方法
 *  join()：所有元素都被处理之前一直阻塞

from threading import Thread, Event from queue import Queue import time import random class producer(Thread): def __init__(self, queue): Thread.__init__(self) self.queue = queue def run(self): for i in range(10): item = random.randint(0, 256) self.queue.put(item) print("Producer notify: item item N° %d appended to queue by %s" % (item, self.name)) time.sleep(1) class consumer(Thread): def __init__(self, queue): Thread.__init__(self) self.queue = queue def run(self): while True: item = self.queue.get() print('Consumer notify : %d popped from queue by %s' % (item, self.name)) self.queue.task_done() if __name__ == "__main__": queue = Queue() t1 = producer(queue) t2 = consumer(queue) t3 = consumer(queue) t4 = consumer(queue) t1.start() t2.start() t3.start() t4.start() t1.join() t2.join() t3.join() t4.join()

## 基于进程的并行 ##

**multiprocessing** 是 Python 标准库中的模块，实现了共享内存机制。

## 异步编程 ##

### 使用 **concurrent.futures** 模块 ###

该模块具有线程池和进程池，管理并行编程任务、处理非确定性的执行流程、进程/线程同步等功能；此模块由以下部分组成

*  concurrent.futures.Executor: 这是一个虚拟基类，提供了异步执行的方法。
 *  submit(function, argument): 调度函数（可调用的对象）的执行，将 argument 作为参数传入。
 *  map(function, argument): 将 argument 作为参数执行函数，以 异步 的方式。
 *  shutdown(Wait=True): 发出让执行者释放所有资源的信号。
 *  concurrent.futures.Future: 其中包括函数的异步执行。Future对象是submit任务（即带有参数的functions）到executor的实例。

示例代码如下所示

import concurrent.futures import time number_list = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] def evaluate_item(x): result_item = count(x) return result_item def count(number): for i in range(0, 1000000): i = i + 1 return i * number if __name__ == "__main__": # 顺序执行 start_time = time.time() for item in number_list: print(evaluate_item(item)) print("Sequential execution in " + str(time.time() - start_time), "seconds") # 线程池执行 start_time_1 = time.time() with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: futures = [executor.submit(evaluate_item, item) for item in number_list] for future in concurrent.futures.as_completed(futures): print(future.result()) print("Thread pool execution in " + str(time.time() - start_time_1), "seconds") # 线程池执行 start_time_2 = time.time() with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor: futures = [executor.submit(evaluate_item, item) for item in number_list] for future in concurrent.futures.as_completed(futures): print(future.result()) print("Process pool execution in " + str(time.time() - start_time_2), "seconds")

#### 使用 Asyncio 管理事件循环 ####

Python 的 Asyncio 模块提供了管理事件、协程、任务和线程的方法，以及编写并发代码的原语。此模块的主要组件和概念包括：

*  事件循环: 在Asyncio模块中，每一个进程都有一个事件循环。
 *  协程: 这是子程序的泛化概念。协程可以在执行期间暂停，这样就可以等待外部的处理（例如IO）完成之后，从之前暂停的地方恢复执行。
 *  Futures: 定义了 Future 对象，和 concurrent.futures 模块一样，表示尚未完成的计算。
 *  Tasks: 这是Asyncio的子类，用于封装和管理并行模式下的协程。

Asyncio 提供了以下方法来管理事件循环：

*  loop = get\_event\_loop(): 得到当前上下文的事件循环。
 *  loop.call\_later(time\_delay, callback, argument): 延后 time\_delay 秒再执行 callback 方法。
 *  loop.call\_soon(callback, argument): 尽可能快调用 callback, call\_soon() 函数结束，主线程回到事件循环之后就会马上调用 callback 。
 *  loop.time(): 以float类型返回当前时间循环的内部时间。
 *  asyncio.set\_event\_loop(): 为当前上下文设置事件循环。
 *  asyncio.new\_event\_loop(): 根据此策略创建一个新的时间循环并返回。
 *  loop.run\_forever(): 在调用 stop() 之前将一直运行。

示例代码如下所示

import asyncio import datetime import time def fuction_1(end_time, loop): print("function_1 called") if(loop.time() + 1.0) < end_time: loop.call_later(1, fuction_2, end_time, loop) else: loop.stop() def fuction_2(end_time, loop): print("function_2 called") if(loop.time() + 1.0) < end_time: loop.call_later(1, function_3, end_time, loop) else: loop.stop() def function_3(end_time, loop): print("function_3 called") if(loop.time() + 1.0) < end_time: loop.call_later(1, fuction_1, end_time, loop) else: loop.stop() def function_4(end_time, loop): print("function_4 called") if(loop.time() + 1.0) < end_time: loop.call_later(1, function_4, end_time, loop) else: loop.stop() loop = asyncio.get_event_loop() end_loop = loop.time() + 9.0 loop.call_soon(fuction_1, end_loop, loop) loop.run_forever() loop.close()

#### 使用 Asyncio 管理协程 ####

示例代码如下所示

import asyncio import time from random import randint @asyncio.coroutine def StartState(): print("Start State called \n") input_val = randint(0, 1) time.sleep(1) if input_val == 0: result = yield from State2(input_val) else: result = yield from State1(input_val) print("Resume of the Transition:\nStart State calling" + result) @asyncio.coroutine def State1(transition_value): outputVal = str("State 1 with transition value=%s \n" % (transition_value)) input_val = randint(0, 1) time.sleep(1) print("...Evaluating...") if input_val == 0: result = yield from State3(input_val) else: result = yield from State2(input_val) @asyncio.coroutine def State2(transition_value): outputVal = str("State 2 with transition value= %s \n" % (transition_value)) input_Val = randint(0, 1) time.sleep(1) print("...Evaluating...") if (input_Val == 0): result = yield from State1(input_Val) else: result = yield from State3(input_Val) result = "State 2 calling " + result return outputVal + str(result) @asyncio.coroutine def State3(transition_value): outputVal = str("State 3 with transition value = %s \n" % (transition_value)) input_val = randint(0, 1) time.sleep(1) print("...Evaluating...") if(input_val == 0): result = yield from State1(input_val) else: result = yield from State2(input_val) result = "State 3 calling " + result return outputVal + str(result) @asyncio.coroutine def EndState(transition_value): outputVal = str("End State With transition value = %s \n" % (transition_value)) print("...Stop Computation...") return outputVal if __name__ == "__main__": print("Finites State Machine simulation with Asyncio Coroutine") loop = asyncio.get_event_loop() loop.run_until_complete(StartState())

#### 使用 Asyncio 控制任务 ####

示例代码如下所示

import asyncio @asyncio.coroutine def factorial(number): f = 1 for i in range(2, number + 1): print("Asyncio.Task:Compute factorial(%s)" % (i)) yield from asyncio.sleep(1) f *= i print("Asyncio.Task - factorial(%s) = %s" % (number, f)) @asyncio.coroutine def fibonacci(number): a, b = 0, 1 for i in range(number): print("Asyncio.Task:Complete fibonacci (%s)" % (i)) yield from asyncio.sleep(1) a, b = b, a+b print("Asyncio.Task - fibonaci (%s)= %s" % (number, a)) @asyncio.coroutine def binomialCoeff(n, k): result = 1 for i in range(1, k+1): result = result * (n-i+1) / i print("Asyncio.Task:Compute binomialCoeff (%s)" % (i)) yield from asyncio.sleep(1) print("Asyncio.Task - binomialCoeff (%s,%s) = %s" % (n, k, result)) if __name__ == "__main__": tasks = [asyncio.Task(factorial(10)), asyncio.Task( fibonacci(10)), asyncio.Task(binomialCoeff(20, 10))] loop = asyncio.get_event_loop() loop.run_until_complete(asyncio.wait(tasks)) loop.close()

#### 使用Asyncio和Futures ####

示例代码如下所示

import asyncio import sys @asyncio.coroutine def first_coroutine(future, N): count = 0 for i in range(1, N + 1): count = count + i yield from asyncio.sleep(4) future.set_result( "first coroutine (sum of N integers) result = " + str(count)) @asyncio.coroutine def second_coroutine(future, N): count = 1 for i in range(2, N + 1): count *= i yield from asyncio.sleep(3) future.set_result("second coroutine (factorial) result = " + str(count)) def got_result(future): print(future.result()) if __name__ == "__main__": N1 = 1 N2 = 1 loop = asyncio.get_event_loop() future1 = asyncio.Future() future2 = asyncio.Future() tasks = [ first_coroutine(future1, N1), second_coroutine(future2, N2) ] future1.add_done_callback(got_result) future2.add_done_callback(got_result) loop.run_until_complete(asyncio.wait(tasks)) loop.close()

## 分布式编程 ##

略

## GPU 编程 ##

略

## 相关参考 ##

*  [Python并行编程 中文版][Python_]

转载于:https://www.cnblogs.com/hippieZhou/p/10205308.html

[Python_]: https://python-parallel-programmning-cookbook.readthedocs.io/zh_CN/latest/index.html