我家小花儿

这个屌丝很懒，什么也没留下！

热门标签

python 并发学习（基于线程的并行）_python-parallel-programning-cookbook.readthedocs.i

作者：我家小花儿 | 2024-08-11 08:27:48

踩

python-parallel-programning-cookbook.readthedocs.io

背景

通常一个应用有一个进程，分成多个独立的线程，并行运行，互相配合，执行不同类型的任务。
线程是独立的处理流程，，可以和系统的其他线程并行或者并发的执行。多线程可以共享资源和数据，利用所谓的共享内存空间。线程和进程的使用取决于你要运行的操作系统。但是总体来说我们可以说线程是包含在进程之间的，统一进程可以有多个不同的线程可以共享相同的资源。想比而言，进程之间不会进行共享资源。

每个线程基本上包含3个元素，程序计数器，寄存器和栈。与同一进程的其他线程共享的资源基本上包含数据和系统资源。每一个线程都有自己的运行状态，可以和其他线程同步，这一点和进程一样。线程的状态答题可以分为 ready running blocked. 线程的典型应用是应用软件的并行化–为了利用现代的多核处理器，使每个核心都可以运行单个线程，相比于进程，使用线程的优势是性能。相比较进程，在进程之间切换上下文要比在统一进程的多线程之间切换上下文要重的多。

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

使用python 的线程模块

Python 通过标准库的threading 模块来管理线程。线程的主要组件：

线程对象
Lock 对象
Rlock 对象
信号对象
条件对象
事件对象

threading 模块提供了Thread () 方法在不同的线程中运行函数或者处理过程

class threading.Thread(group=None,
                       target=None,
                       name=None,
                       args=(),
                       kwargs={})1
2
3
4
5

group 一般为None 为以后得特性留的
target 当线程启动的时候要执行函数
name 线程的名字，默认会分配一个唯一的名字 Thread-N
args 传递给target 的参数使用tuple类型
kwargs: 同上，使用字段类型dict

创建线程的方法非常使用，通过参数和目标告诉线程应该怎么做。

下面看代码：

import threading


def function(i):
    print("function called by thread {}".format(i))
    return


threads = []
for i in range(10):
    t = threading.Thread(target=function, args=(i,))
    threads.append(t)
    t.start()
    t.join()1
2
3
4
5
6
7
8
9
10
11
12
13
14
'运行

线程被创建之后并不会马上执行，还需要手动调用start(), join() 让调用它的线程一直等到执行结束（即阻塞他的主线程， t线程执行结束，主线程才会继续执行）

t.start()
t.join()1
2

如何确定当前线程

使用参数来确认或命名线程是笨拙且没有必要的。每一个 Thread 实例创建的时候都有一个带默认值的名字，并且可以修改。在服务端通常一个服务进程都有多个线程服务，负责不同的操作，这时候命名线程是很实用的。

为了演示线程的执行我们创建3个目标函函数，引入time在运行时候挂起2秒

import threading
import time

def first_function():
    print(threading.currentThread().getName() + str(' is Starting '))
    time.sleep(2)
    print (threading.currentThread().getName() + str(' is Exiting '))
    return

def second_function():
    print(threading.currentThread().getName() + str(' is Starting '))
    time.sleep(2)
    print (threading.currentThread().getName() + str(' is Exiting '))
    return

def third_function():
    print(threading.currentThread().getName() + str(' is Starting '))
    time.sleep(2)
    print(threading.currentThread().getName() + str(' is Exiting '))
    return

if __name__ == "__main__":
    t1 = threading.Thread(name='first_function', target=first_function)
    t2 = threading.Thread(name='second_function', target=second_function)
    t3 = threading.Thread(name='third_function', target=third_function)
    t1.start()
    t2.start()
    t3.start()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
'运行

运行结果

first_function is Starting 
second_function is Starting 
third_function is Starting 
first_function is Exiting 
second_function is Exiting 
third_function is Exiting 1
2
3
4
5
6

可以看到前3 个都是starting 都是互不影响的几乎都是同时执行的，各自阻挡各自得线程2秒。互不影响。

实现一个线程

threading模块实现一个新的线程，需要下面3步：

定义一个 Thread 类的子类
覆盖 init(self [,args]) 方法，可以添加额外的参数
最后，需要覆盖 run(self, [,args]) 方法来实现线程要做的事情

当你创建了新的 Thread 子类的时候，你可以实例化这个类，调用 start() 方法来启动它。线程启动之后将会执行 run() 方法

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 " + self.name)
        print_time(self.name, self.counter, 5)
        print("Exiting " + self.name)

def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            import _thread
            _thread.exit()
        time.sleep(delay)
        print("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1

# Create new threads
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# Start new Threads
thread1.start()
thread2.start()

thread1.join()
thread2.join()
print("Exiting Main Thread")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

总结
线程模块是创建和管理线程的首选形式。每一个线程都通过一个集成Thread 的子类代表，覆盖run() 方法来实现逻辑，这个方法是线程的入口

在主程序中我们创建多个myThread 的类型实例，然后执行start()方法启动。调用Thread 的构造器是必须的。通过他我们可以给线程定义一些名字或者分类之类的属性。调用start()之后哦线变成活跃状态，并且直到持续run() 结束，或者中间出现异常。所有的线程执行完了之后，程序结束。
join() 命令控制主线程的终止

也就是和java 中的一样耗时的操作我们的都放在重写的run（）方法中去

使用Lock进行线程同步

当两个或者以上对共享内存的操作发生在并发线程中的时候，并且至少有一个可以改变数据，又没有同步机制的作用下，就会产生竞争的条件，可能会导致执行无效的代码。bug 或者异常

竞争条件最简单的解决办法是使用锁。锁的操作非常简单，当一个线程需要访问部分共享内存的时候，必须先获得锁才能访问。此线程对这部分共享资源使用完成之后，该线程必须释放锁。然后其他线程就可以拿到这个锁并访问到这部分的资源啦。

很显然，避免竞争条件出现是非常重要的，所以我们要保证，在同一时刻只有一个线程允许访问共享内存。
尽管原理很简单，但是这样使用是work的。

在实际使用的过程中，我们发现这个方法经常会导致一种糟糕的死锁现象。当不同的线程要求得到一个锁时，死锁就会发生，这时程序不可能继续执行，因为它们互相拿着对方需要的锁

这里写图片描述

为了简化问题，我们设有两个并发的线程（线程A 和线程B )，需要资源1 和资源2 .假设线程A 需要资源1 ，线程B 需要资源2 .在这种情况下，两个线程都使用各自的锁，目前为止没有冲突。现在假设，在双方释放锁之前，线程A 需要资源2 的锁，线程B 需要资源1 的锁，没有资源线程不会继续执行。鉴于目前两个资源的锁都是被占用的，而且在对方的锁释放之前都处于等待且不释放锁的状态。这是死锁的典型情况。所以如上所说，使用锁来解决同步问题是一个可行却存在潜在问题的方案。

本节中，我们描述了Python的线程同步机制， lock() 。通过它我们可以将共享资源某一时刻的访问限制在单一线程或单一类型的线程上，线程必须得到锁才能使用资源，并且之后必须允许其他线程使用相同的资源。

如何做

下面的例子展示了如何通过 lock() 管理线程。在下面的代码中，我们有两个函数： increment() 和 decrement() 。第一个函数对共享资源执行加1的操作，另一个函数执行减1.两个函数分别使用线程封装。除此之外，每一个函数都有一个循环重复执行操作。我们想要保证，通过对共享资源的管理，执行结果是共享资源最后等于初始值0.

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

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_without_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_without_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 %s" % shared_resource_with_lock)
    print("the value of shared variable with race condition is %s" % shared_resource_with_no_lock)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
'运行

结果可能每次I都不一样

the value of shared variable with lock management is 0
the value of shared variable with race condition is 100000
1
2
3

可以看到有锁的会一直是 0 而没有锁的话使用共享资源就会出现问题
在 increment_with_lock() 函数和 decrement_with_lock() 函数中，可以看到我们使用了lock语句。当你需要使用资源的时候，调用 acquire() 拿到锁（如果锁暂时不可用，会一直等待直到拿到），最后调用 release():

shared_resource_lock.acquire()
shared_resource_with_lock -= 1
shared_resource_lock.release()
1
2
3

总结
锁有两种状态： locked（被某一线程拿到）和unlocked（可用状态）
我们有两个方法来操作锁： acquire() 和 release()

需要遵循以下规则：
如果状态是unlocked，可以调用 acquire() 将状态改为locked
如果状态是locked， acquire() 会被block直到另一线程调用 release() 释放锁
如果状态是unlocked，调用 release() 将导致 RuntimError 异常
如果状态是locked，可以调用 release() 将状态改为unlocked

注意啦
尽管理论上行得通，但是锁的策略不仅会导致有害的僵持局面。还会对应用程序的其他方面产生负面影响。这是一种保守的方法，经常会引起不必要的开销，也会限制程序的可扩展性和可读性。更重要的是，有时候需要对多进程共享的内存分配优先级，使用锁可能和这种优先级冲突。最后，从实践的经验来看，使用锁的应用将对debug带来不小的麻烦。所以，最好使用其他可选的方法确保同步读取共享内存，避免竞争条件。

使用RLock进行线程同步

如果你想让只有拿到锁的线程才能释放该锁，那么应该使用 RLock() 对象。和 Lock() 对象一样， RLock() 对象有两个方法： acquire() 和 release() 。当你需要在类外面保证线程安全，又要在类内使用同样方法的时候 RLock() 就很实用了。

（译者注：RLock原作解释的太模糊了，译者在此擅自添加一段。RLock其实叫做“Reentrant Lock”，就是可以重复进入的锁，也叫做“递归锁”。这种锁对比Lock有是三个特点：1. 谁拿到谁释放。如果线程A拿到锁，线程B无法释放这个锁，只有A可以释放；2. 同一线程可以多次拿到该锁，即可以acquire多次；3. acquire多少次就必须release多少次，只有最后一次release才能改变RLock的状态为unlocked）

在示例代码中，我们引入了 Box 类，有 add() 方法和 remove() 方法，提供了进入 execute() 方法的入口。 execute() 的执行由 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()

## These two functions run n in separate
## threads and call the Box's methods
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

## the main program build some
## threads and make sure it works
if __name__ == "__main__":
    items = 5
    print("putting %s items in the box " % 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("%s items still remain in the box " % box.total_items)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
'运行

运行结果

putting 5 items in the box 
adding 1 item in the box
removing 1 item in the box
adding 1 item in the box
removing 1 item in the box
adding 1 item in the box
removing 1 item in the box
adding 1 item in the box
removing 1 item in the box
adding 1 item in the box
removing 1 item in the box
0 items still remain in the box 1
2
3
4
5
6
7
8
9
10
11
12

adder() 和 remover() 两个函数在 Box 类内操作items，即调用 Box 类的方法： add() 和 remove() 。每一次方法调用，都会有一次拿到资源然后释放资源的过程。至于 lock() 对象， RLock() 对象有 acquire() 和 release() 方法可以拿到或释放资源；然后每一次方法调用中，我们都有以下操作：

Box.lock.acquire()
# ...do something
Box.lock.release()
1
2
3
4

使用信号量进行线程同步

信号量是由操作系统管理的一种抽象数据类型，用于在多线程中同步对共享资源的使用。
本质上来说信号量是一个内部数据，用于标明当前的共享资源可以有多少并发读取。
信号量的操作有两个函数： acquire() 和release()

每当线程想要读取关联了信号量的共享资源时候，必须调用acquire(), 此操作减少信号量的内部变量，如果此变量的值非负，那么分配该资源的权限。如果是负值则线程被挂起，直到有其他的线程释放资源。
当线程不在需要该共享资源，必须通过release（）释放。这样信号量的内部变量增加，在信号量等待队列中排在最前面的线程会拿到共享资源权限

虽然表面上看信号量机制没什么明显的问题，如果信号量的等待和通知操作都是原子的，确实没什么问题。但如果不是，或者两个操作有一个终止了，就会导致糟糕的情况。

举个例子，假设有两个并发的线程，都在等待一个信号量，目前信号量的内部值为1。假设第线程A将信号量的值从1减到0，这时候控制权切换到了线程B，线程B将信号量的值从0减到-1，并且在这里被挂起等待，这时控制权回到线程A，信号量已经成为了负值，于是第一个线程也在等待。

这样的话，尽管当时的信号量是可以让线程访问资源的，但是因为非原子操作导致了所有的线程都在等待状态。

准备工作

下面的代码展示了信号量的使用，我们有两个线程， producer() 和 consumer() ，它们使用共同的资源，即item。 producer() 的任务是生产item， consumer() 的任务是消费item。

当item还没有被生产出来， consumer() 一直等待，当item生产出来， producer() 线程通知消费者资源可以使用了。

在以下的代码中，我们使用生产者-消费者模型展示通过信号量的同步。当生产者生产出item，便释放信号量。然后消费者拿到资源进行消费。


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

"""Using a Semaphore to synchronize threads"""
import threading
import time
import random

# The optional argument gives the initial value for the internal
# counter;
# it defaults to 1.
# If the value given is less than 0, ValueError is raised.
semaphore = threading.Semaphore(0)


def consumer():
    print("consumer is waiting.")
    # Acquire a semaphore
    semaphore.acquire()
    # The consumer have access to the shared resource
    print("Consumer notify : consumed item number %s " % item)


def producer():
    global item
    time.sleep(3)
    # create a random item
    item = random.randint(0, 1000)
    print("producer notify : produced item number %s" % item)
    # Release a semaphore, incrementing the internal counter by one.
    # When it is zero on entry and another thread is waiting for it
    # to become larger than zero again, wake up that thread.
    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")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

运行结果

consumer is waiting.
producer notify : produced item number 553
Consumer notify : consumed item number 553 
consumer is waiting.
producer notify : produced item number 548
Consumer notify : consumed item number 548 
consumer is waiting.
producer notify : produced item number 116
Consumer notify : consumed item number 116 
consumer is waiting.
producer notify : produced item number 420
Consumer notify : consumed item number 420 
consumer is waiting.
producer notify : produced item number 535
Consumer notify : consumed item number 535 
program terminated
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

通过结果分析信号量的使用原理

consumer is waiting.
producer notify : produced item number 553
Consumer notify : consumed item number 553 
1
2
3

生产者和消费者都是两个不同的线程产生的，生产者需要等3秒往下执行然后去释放资源，这里的原则是不是岁锁的谁来释放，有消费者来锁，由生产者来释放资源。 consumer is waiting. 之后遇到资源 semaphore.acquire() 之后消费者线程被挂起不执行啦，相当于停在这里啦，等到sleep 醒来之后生产者释放资源信号量释放。然后消费者的线程继续往下跑
信号量被初始化为0，此信号量唯一的目的是同步两个或多个线程。在这里，我们的线程必须并行运行，所以需要信号量同步：

semaphore = threading.Semaphore(0)
1

这个操作和lock中的机制非常相似， producer() 完成创建item之后，释放资源：

semaphore.release()
1

信号量的 release() 可以提高计数器然后通知其他的线程。同样的， consumer() 方法可以通过下面的方法拿到资源：

semaphore.acquire()
1

如果信号量的计数器到了0，就会阻塞 acquire() 方法，直到得到另一个线程的通知。如果信号量的计数器大于0，就会对这个值-1然后分配资源。
最后，拿到数据并打印输出：

print("Consumer notify : consumed item number %s " % item)
1

了解更多

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

信号量在支持多线程的编程语言中依然应用很广，然而这可能导致死锁的情况。例如，现在有一个线程t1先等待信号量s1，然后等待信号量s2，而线程t2会先等待信号量s2，然后再等待信号量s1，这样就可能会发生死锁，导致t1等待s2，但是t2在等待s1。

使用条件进行线程同步

条件值得是应用程序状态的改变。这是另外一种同步机制，其中某些线程在等待某一条件发生，其他的线程会在该条件发生的时候进行通知，一旦条件发生，线程会拿到共享资源的唯一权限。

准备条件

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

代码：

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 consume")
        items.pop()
        print("Consumer notify : consumed 1 item")
        print("Consumer notify : items to consume are " + str(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 " + str(len(items)))
            print("Producer notify : stop the production!!")
        items.append(1)
        print("Producer notify : total items producted " + str(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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

运行结果

Producer notify : total items producted 1
Consumer notify : consumed 1 item
Consumer notify : items to consume are 0
Producer notify : total items producted 1
Producer notify : total items producted 2
Consumer notify : consumed 1 item
Consumer notify : items to consume are 1
Producer notify : total items producted 2
Producer notify : total items producted 3
Consumer notify : consumed 1 item
Consumer notify : items to consume are 2
Producer notify : total items producted 3
Producer notify : total items producted 4
Consumer notify : consumed 1 item
Consumer notify : items to consume are 3
Producer notify : total items producted 4
Producer notify : total items producted 5
Consumer notify : consumed 1 item
Consumer notify : items to consume are 4
Producer notify : total items producted 5
Producer notify : total items producted 6
Consumer notify : consumed 1 item
Consumer notify : items to consume are 5
Producer notify : total items producted 6
Producer notify : total items producted 7
Consumer notify : consumed 1 item
Consumer notify : items to consume are 6
Producer notify : total items producted 7
Producer notify : total items producted 8
Consumer notify : consumed 1 item
Consumer notify : items to consume are 7
Producer notify : total items producted 8
Producer notify : total items producted 9
Consumer notify : consumed 1 item
Consumer notify : items to consume are 8
Producer notify : total items producted 9
Producer notify : total items producted 10
Consumer notify : consumed 1 item
Consumer notify : items to consume are 9
Producer notify : total items producted 10
Consumer notify : consumed 1 item
Consumer notify : items to consume are 9
Consumer notify : consumed 1 item
Consumer notify : items to consume are 8
Consumer notify : consumed 1 item
Consumer notify : items to consume are 7
Consumer notify : consumed 1 item
Consumer notify : items to consume are 6
Consumer notify : consumed 1 item
Consumer notify : items to consume are 5
Consumer notify : consumed 1 item
Consumer notify : items to consume are 4
Consumer notify : consumed 1 item
Consumer notify : items to consume are 3
Consumer notify : consumed 1 item
Consumer notify : items to consume are 2
Consumer notify : consumed 1 item
Consumer notify : items to consume are 1
Consumer notify : consumed 1 item
Consumer notify : items to consume are 0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60

分享
消费者通过拿到锁来修改共享的资源 items[] ：

condition.acquire()
1

如果list的长度为0，那么消费者就进入等待状态

if len(items) == 0:
condition.wait()
1
2

否则就通过 pop 操作消费一个item：

items.pop()
1

然后，消费者的状态被通知给生产者，同时共享资源释放：

condition.notify()
condition.release()
1
2

生产者拿到共享资源，然后确认缓冲队列是否已满（在我们的这个例子中，最大可以存放10个item），如果已经满了，那么生产者进入等待状态，知道被唤醒：

condition.acquire()
if len(items) == 10:
    condition.wait()
1
2
3

如果队列没有满，就生产1个item，通知状态并释放资源：

condition.notify()
condition.release()
1
2

了解更多

Python对条件同步的实现很有趣。如果没有已经存在的锁传给构造器的话，内部的 _Condition 会创建一个 RLock() 对象。同时，这个RLock也会通过 acquire() 和 release() 管理：

class _Condition(_Verbose):
    def __init__(self, lock=None, verbose=None):
        _Verbose.__init__(self, verbose)
        if lock is None:
           lock = RLock()
        self.__lock = lock1
2
3
4
5
6

哈哈哈怎么都感觉越来越向yield 靠近啦
感觉线程多的时候就不知道怎么判断啦

使用事件进行线程同步

事件是线程之间用于通讯对象。有的线程等待信号有的线程发出信号。基本上事件对象都会维护一个内部变量，可以通过set() 方法设置为true, 也可以通过clear() 方法设置为false,. wait() 方法将会阻塞线程，直到内部变量变为true.

如何做

为了理解事件对象实现的线程同步，再一次看生产者消费者问题

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

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 : %d popped from list by %s' % (item, self.name))


class producer(Thread):
    """生产者"""

    def __init__(self, items, event):
        Thread.__init__(self)
        self.items = items
        self.event = event

    def run(self):
        global item
        for i in range(10):
            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)

    t2.start()
    t1.start()
    t2.join()
    t1.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

python 线程事件用于主线程控制其他线程执行，事件主要方法： set() clear() wait()
事件处理的机制：
全局定义了一个‘Flag’ 如果Flag为False, 那么当程序执行event .wait()方法时候就会阻塞，如果Flag 的值为True , 那么event.wait() 方法时候便不在进行阻塞。
clear: 将Flag 设置为False,
set : 将Flag 设置为True

用threading.Event 实现线程间通信，使用threading.Event 可以使用一个线程等待另外一个线程的通知，我们把这个event传递到线程中， Event 默认了一个标志 False，一旦线程通过wait() 方法进入到等待状态，直到另外一个线程调用该Event 的set方法将内置标志设置为True 时，该Event 会通知所有处于等待状态的线程恢复运行。

感兴趣的同学可以执行下代码查看打印。再对比这个机制就会理解事件

class consumer(Thread):
def __init__(self, items, event):
    Thread.__init__(self)
    self.items = items
    self.event = event
1
2
3
4
5

producer 类初始化时定义了item的list和 Event ，与条件对象时候的例子不同，这里的list并不是全局的，而是通过参数传入的

使用with语法

Python从2.5版本开始引入了 with 语法。此语法非常实用，在有两个相关的操作需要在一部分代码块前后分别执行的时候，可以使用 with 语法自动完成。同事，使用 with 语法可以在特定的地方分配和释放资源，因此， with 语法也叫做“上下文管理器”。在threading模块中，所有带有 acquire() 方法和 release() 方法的对象都可以使用上下文管理器。

Lock
RLock
Condition
Semaphore

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__':
    # let's create a test battery
    lock = threading.Lock()
    rlock = threading.RLock()
    condition = threading.Condition()
    mutex = threading.Semaphore(1)
    threading_synchronization_list = [lock, rlock, condition, mutex]
    # in the for cycle we call the threading_with e threading_no_with function
    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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
'运行

使用了Python的logging模块进行输出：

logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)
1

使用 % (threadName) 可以在每次输出的信息都加上线程的名字。logging模块是线程安全的。这样我们可以区分出不同线程的输出。

使用 queue 进行线程通信

前面我们已经讨论到，当线程之间如果要共享资源或数据的时候，可能变的非常复杂。如你所见，Python的threading模块提供了很多同步原语，包括信号量，条件变量，事件和锁。如果可以使用这些原语的话，应该优先考虑使用这些，而不是使用queue（队列）模块。队列操作起来更容易，也使多线程编程更安全，因为队列可以将资源的使用通过单线程进行完全控制，并且允许使用更加整洁和可读性更高的设计模式。

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

put(): 往queue中放一个item
get(): 从queue删除一个item，并返回删除的这个item
task_done(): 每次item被处理的时候需要调用这个方法
join(): 所有item都被处理之前一直阻塞

本例中将会有两个实体视图共享临界资源，一个队列。

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 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()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

生产者使用 Queue.put(item [,block[, timeout]]) 来往queue中插入数据。Queue是同步的，在插入数据之前内部有一个内置的锁机制。
可能发生两种情况：

如果 block 为 True ， timeout 为 None （这也是默认的选项，本例中使用默认选项），那么可能会阻塞掉，直到出现可用的位置。如果 timeout 是正整数，那么阻塞直到这个时间，就会抛出一个异常。
如果 block 为 False ，如果队列有闲置那么会立即插入，否则就立即抛出异常（ timeout 将会被忽略）。本例中， put() 检查队列是否已满，然后调用 wait() 开始等待。
消费者从队列中取出整数然后用 task_done() 方法将其标为任务已处理。

消费者使用 Queue.get([block[, timeout]]) 从队列中取回数据，queue内部也会经过锁的处理。如果队列为空，消费者阻塞。

这里写图片描述

借鉴文章

http://python-parallel-programmning-cookbook.readthedocs.io/zh_CN/latest/chapter1/index.html

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家小花儿/article/detail/963164

python 并发学习（基于线程的并行）_python-parallel-programning-cookbook.readthedocs.i

背景

使用python 的线程模块

如何确定当前线程

实现一个线程

使用Lock进行线程同步

如何做

使用RLock进行线程同步

使用信号量进行线程同步

准备工作

使用条件进行线程同步

准备条件

了解更多

使用事件进行 线程同步

如何做

使用with语法

使用 queue 进行线程通信

借鉴文章

使用事件进行线程同步