并行编程

中文版》https://python-parallel-programmning-cookbook.read***docs.io/zh_CN/latest/index.html

线程和进程

进程可以包含多个并行运行的线程通常操作系统创建和管理线程比进程更省CPU资源线程用于一些小任务进程用于繁重的任务同一个进程下的线程共享地址空间和其他资源进程之间相互独立

在Python中使用线程:

线程是独立的处理流程可以和系统的其他线程并行或并发地执行。

多线程可以共享数据和资源利用所谓的共享内存空间。

每一个线程基本上包含3个元素程序计数器寄存器和栈。

线程的状态大体上可以分为ready,

running,

blocked。

多线程编程一般使用共享内容空间进行线程间的通讯这就使管理内容空间成为多线程编程的重点和难点。

线程的典型应用是应用软件的并行化。

相比于进程使用线程的优势主要是性能。

2.2

threading.Thread(targetfunction,

args(i,))

threading.current_thread().name

通常一个服务进程中有多个线程服务负责不同的操作所以对于线程的命名是很重要的Python中每一个线程在被

Thread

first_func():print(threading.current_thread().name

str(

Starting))time.sleep(2)print(threading.current_thread().name

str(is

second_func():print(threading.current_thread().name

str(

Starting))time.sleep(2)print(threading.current_thread().name

str(is

third_func():print(threading.current_thread().name

str(

Starting))time.sleep(2)print(threading.current_thread().name

str(is

threading.Thread(namefirst_func,

targetfirst_func)t2

threading.Thread(namesecond_func,

targetsecond_func)t3

threading.Thread(targetthird_func)t1.start()t2.start()t3.start()t1.join()t2.join()t3.join()运行结果

first_func

方法并添加额外的参数threading.Thread.__init__(self)

初始化继承自Thread类的属性使子类对象能够正确地继承和使用父类的属性和方法self.threadID

threadID

self.name)print_time(self.name,

self.counter,

_thread.exit()time.sleep(delay)print(%s:

(threadName,

time.ctime(time.time())))counter

1##

并发线程中多个线程对共享内存进行操作并且至少有一个可以改变数据。

这种情况下如果没有同步机制那么多个线程之间就会产生竞争从而导致代码无效或出错。

解决多线程竞争问题的最简单的方法就是用锁

Lock

进行访问资源。

因此为了避免多线程竞争的出现必须保证同一时刻只能允许一个线程访问共享内存。

在实际使用中该方法经常会导致一种

详见https://python-parallel-programmning-cookbook.read***docs.io/zh_CN/latest/chapter2/06_Thread_synchronization_with_Lock_and_Rlock.html

举例:

threadingshared_resource_with_lock

COUNT

shared_resource_with_lock这样写就不需要再通过函数的参数引入

range(COUNT):shared_resource_lock.acquire()

1shared_resource_lock.release()

锁def

range(COUNT):shared_resource_lock.acquire()shared_resource_with_lock

1shared_resource_lock.release()##

def

increment_without_lock():global

shared_resource_with_no_lockfor

range(COUNT):shared_resource_with_no_lock

1def

decrement_without_lock():global

shared_resource_with_no_lockfor

range(COUNT):shared_resource_with_no_lock

1if

threading.Thread(targetincrement_with_lock)t2

threading.Thread(targetdecrement_with_lock)t3

threading.Thread(targetincrement_without_lock)t4

threading.Thread(targetdecrement_without_lock)##

开启线程t1.start()t2.start()t3.start()t4.start()##

.join()t1.join()t2.join()t3.join()t4.join()print

(***

shared_resource_with_lock)print

(***

shared_resource_with_no_lock)运行结果

***

0尽管在上面的结果中没锁的情况下得到的结果也是正确的但是执行多次总会出现错误的结果而有锁的情况下执行多次结果一定是正确的。

尽管理论上用锁的策略可以避免多线程中的竞争问题但是可能会对程序的其他方面产生负面影响。

此外锁的策略经常会导致不必要的开销也会限制程序的可扩展性和可读性。

更重要的是有时候需要对多进程共享的内存分配优先级使用锁可能和这种优先级冲突。

从实践的经验来看使用锁的应用将对debug带来不小的麻烦。

所以最好使用其他可选的方法确保同步读取共享内存避免竞争条件。

使用RLock进行线程同步:

Lock()一样RLock()也有acquire()和release()两种方法RLock()

有三个特点

谁拿到谁释放。

如果线程A拿到锁线程B无法释放这个锁只有A可以释放

2).

acquire多少次就必须release多少次只有最后一次release才能改变RLock的状态为unlocked;

import

__init__(self):self.total_items

0def

n):Box.lock.acquire()self.total_items

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,

box)box.add()time.sleep(1)items

1def

box)box.remove()time.sleep(1)items

1if

threading.Thread(targetremover,

args(box,

items))t1.start()t2.start()t1.join()t2.join()print(%s

items

Box类的execute()方法包含RLockadder()和remover()方法也包含RLock就是说无论是调用Box还是adder()或者remover()每个线程的每一步都有拿到资源、释放资源的过程。

使用信号量进行线程同步

信号量是由操作系统管理的一种抽象数据类型用于多线程中同步对共享资源的使用信号量是一个内部数据用于表明当前共享资源可以有多少并发读取在

Threading

中信号量的操作有两个函数acquire()和release()

import

waiting.)semaphore.acquire()print(consumer

notify:

item)semaphore.release()print(producer

semaphore

threading.Thread(targetproducer)t2

threading.Thread(targetconsumer)t1.start()t2.start()t1.join()t2.join()print(program

terminated)

threading.Semaphore(0)将信号量初始化为0其目的在于同步两个或多个线程。

producer()执行完后通过seaphore.release()释放资源之后consumer()通过semaphore.acquire()拿到资源

信号量的一个特殊用法是互斥量互斥量是初始值为1的信号量可以实现数据、资源的互斥访问

使用条件进行线程同步

条件指的是程序状态的改变某些线程在等待某一条件发生其他的线程会在该条件发生的时候进行通知。

一旦条件发生线程会拿到共享资源的唯一权限。

from

__init__(self):Thread.__init__(self)def

consume(self):global

consume)items.pop()print(Consumer

notify

str(len(items)))condition.notify()

Wake

20):time.sleep(2)self.consume()class

producer(Thread):def

__init__(self):Thread.__init__(self)def

produce(self):global

10:condition.wait()print(Producer

notify

production!!)items.append(1)print(Producer

notify

str(len(items)))condition.notify()condition.release()def

run(self):for

20):time.sleep(1)self.produce()if

__name__

consumer()producer.start()consumer.start()producer.join()consumer.join()运行结果

Producer

(https://blog.csdn.net/lzanze/article/details/105351064)

使用事件进行线程同步

event):Thread.__init__(self)self.items

itemsself.event

True:time.sleep(2)self.event.wait()

线程会阻塞在这里直到事件

event):Thread.__init__(self)self.items

itemsself.event

self.event.is_set())self.event.clear()

clear()

event)t1.start()t2.start()t1.join()t2.join()运行结果部分结果

Producer

Thread-64该脚本我自己在运行的时候运行了20多分钟没结束所以直接停了部分结果如上所示。

with

logging.basicConfig(levellogging.DEBUG,

%(message)s,

threading_not_with(statement):statement.acquire()try:logging.debug(%s

acquired

statement)finally:statement.release()if

__name__

threading.Semaphore(1)threading_synchronization_list

[lock,

threading_synchronization_list:t1

threading.Thread(targetthreading_with,

args(statement,))t2

threading.Thread(targetthreading_not_with,

args(statement,))t1.start()t2.start()t1.join()t2.join()运行结果

(Thread-68

directly上述结果展示了使用with和不用with的每一个函数以及用在了哪些地方(Lock,

RLock,