英文出处: Akshar Raaj。欢迎加入 翻译组。
我们将使用Python线程来解决Python中的生产者—消费者问题。这个问题完全不像他们在学校中说的那么难。
如果你对生产者—消费者问题有了解,看这篇博客会更有意义。
为什么要关心生产者—消费者问题:
- 可以帮你更好地理解并发和不同概念的并发。
- 信息队列中的实现中,一定程度上使用了生产者—消费者问题的概念,而你某些时候必然会用到消息队列。
当我们在使用线程时,你可以学习以下的线程概念:
- Condition:线程中的条件。
- wait():在条件实例中可用的wait()。
- notify() :在条件实例中可用的notify()。
我假设你已经有这些基本概念:线程、竞态条件,以及如何解决静态条件(例如使用lock)。否则的话,你建议你去看我上一篇文章basics of Threads。
引用维基百科:
生产者的工作是产生一块数据,放到buffer中,如此循环。与此同时,消费者在消耗这些数据(例如从buffer中把它们移除),每次一块。
这里的关键词是“同时”。所以生产者和消费者是并发运行的,我们需要对生产者和消费者做线程分离。
1
2
3
4
5
6
7
8
9
|
from
threading
import
Thread
class
ProducerThread
(
Thread
)
:
def
run
(
self
)
:
pass
class
ConsumerThread
(
Thread
)
:
def
run
(
self
)
:
pass
|
再次引用维基百科:
这个为描述了两个共享固定大小缓冲队列的进程,即生产者和消费者。
假设我们有一个全局变量,可以被生产者和消费者线程修改。生产者产生数据并把它加入到队列。消费者消耗这些数据(例如把它移出)。
1
|
queue
=
[
]
|
在刚开始,我们不会设置固定大小的条件,而在实际运行时加入(指下述例子)。
一开始带bug的程序:
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
|
from
threading
import
Thread
,
Lock
import
time
import
random
queue
=
[
]
lock
=
Lock
(
)
class
ProducerThread
(
Thread
)
:
def
run
(
self
)
:
nums
=
range
(
5
)
#Will create the list [0, 1, 2, 3, 4]
global
queue
while
True
:
num
=
random
.
choice
(
nums
)
#Selects a random number from list [0, 1, 2, 3, 4]
lock
.
acquire
(
)
queue
.
append
(
num
)
print
"Produced"
,
num
lock
.
release
(
)
time
.
sleep
(
random
.
random
(
)
)
class
ConsumerThread
(
Thread
)
:
def
run
(
self
)
:
global
queue
while
True
:
lock
.
acquire
(
)
if
not
queue
:
print
"Nothing in queue, but consumer will try to consume"
num
=
queue
.
pop
(
0
)
print
"Consumed"
,
num
lock
.
release
(
)
time
.
sleep
(
random
.
random
(
)
)
ProducerThread
(
)
.
start
(
)
ConsumerThread
(
)
.
start
(
)
|
运行几次并留意一下结果。如果程序在IndexError异常后并没有自动结束,用Ctrl+Z结束运行。
样例输出:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
Produced
3
Consumed
3
Produced
4
Consumed
4
Produced
1
Consumed
1
Nothing
in
queue
,
but
consumer
will
try
to
consume
Exception
in
thread
Thread
-
2
:
Traceback
(
most
recent
call
last
)
:
File
"/usr/lib/python2.7/threading.py"
,
line
551
,
in
__bootstrap_inner
self
.
run
(
)
File
"producer_consumer.py"
,
line
31
,
in
run
num
=
queue
.
pop
(
0
)
IndexError
:
pop
from
empty
list
|
解释:
- 我们开始了一个生产者线程(下称生产者)和一个消费者线程(下称消费者)。
- 生产者不停地添加(数据)到队列,而消费者不停地消耗。
- 由于队列是一个共享变量,我们把它放到lock程序块内,以防发生竞态条件。
- 在某一时间点,消费者把所有东西消耗完毕而生产者还在挂起(sleep)。消费者尝试继续进行消耗,但此时队列为空,出现IndexError异常。
- 在每次运行过程中,在发生IndexError异常之前,你会看到print语句输出”Nothing in queue, but consumer will try to consume”,这是你出错的原因。
我们把这个实现作为错误行为(wrong behavior)。
什么是正确行为?
当队列中没有任何数据的时候,消费者应该停止运行并等待(wait),而不是继续尝试进行消耗。而当生产者在队列中加入数据之后,应该有一个渠道去告诉(notify)消费者。然后消费者可以再次从队列中进行消耗,而IndexError不再出现。
关于条件
- 条件(condition)可以让一个或多个线程进入wait,直到被其他线程notify。参考:?http://docs.python.org/2/library/threading.html#condition-objects
这就是我们所需要的。我们希望消费者在队列为空的时候wait,只有在被生产者notify后恢复。生产者只有在往队列中加入数据后进行notify。因此在生产者notify后,可以确保队列非空,因此消费者消费时不会出现异常。
- condition内含lock。
- condition有acquire()和release()方法,用以调用内部的lock的对应方法。
condition的acquire()和release()方法内部调用了lock的acquire()和release()。所以我们可以用condiction实例取代lock实例,但lock的行为不会改变。
生产者和消费者需要使用同一个condition实例, 保证wait和notify正常工作。
重写消费者代码:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
|
from
threading
import
Condition
condition
=
Condition
(
)
class
ConsumerThread
(
Thread
)
:
def
run
(
self
)
:
global
queue
while
True
:
condition
.
acquire
(
)
if
not
queue
:
print
"Nothing in queue, consumer is waiting"
condition
.
wait
(
)
print
"Producer added something to queue and notified the consumer"
num
=
queue
.
pop
(
0
)
print
"Consumed"
,
num
condition
.
release
(
)
time
.
sleep
(
random
.
random
(
)
)
|
重写生产者代码:
1
2
3
4
5
6
7
8
9
10
11
12
|
class
ProducerThread
(
Thread
)
:
def
run
(
self
)
:
nums
=
range
(
5
)
global
queue
while
True
:
condition
.
acquire
(
)
num
=
random
.
choice
(
nums
)
queue
.
append
(
num
)
print
"Produced"
,
num
condition
.
notify
(
)
condition
.
release
(
)
time
.
sleep
(
random
.
random
(
)
)
|
样例输出:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
Produced
3
Consumed
3
Produced
1
Consumed
1
Produced
4
Consumed
4
Produced
3
Consumed
3
Nothing
in
queue
,
consumer
is
waiting
Produced
2
Producer
added
something
to
queue
and
notified
the
consumer
Consumed
2
Nothing
in
queue
,
consumer
is
waiting
Produced
2
Producer
added
something
to
queue
and
notified
the
consumer
Consumed
2
Nothing
in
queue
,
consumer
is
waiting
Produced
3
Producer
added
something
to
queue
and
notified
the
consumer
Consumed
3
Produced
4
Consumed
4
Produced
1
Consumed
1
|
解释:
- 对于消费者,在消费前检查队列是否为空。
- 如果为空,调用condition实例的wait()方法。
- 消费者进入wait(),同时释放所持有的lock。
- 除非被notify,否则它不会运行。
- 生产者可以acquire这个lock,因为它已经被消费者release。
- 当调用了condition的notify()方法后,消费者被唤醒,但唤醒不意味着它可以开始运行。
- notify()并不释放lock,调用notify()后,lock依然被生产者所持有。
- 生产者通过condition.release()显式释放lock。
- 消费者再次开始运行,现在它可以得到队列中的数据而不会出现IndexError异常。
为队列增加大小限制
生产者不能向一个满队列继续加入数据。
它可以用以下方式来实现:
- 在加入数据前,生产者检查队列是否为满。
- 如果不为满,生产者可以继续正常流程。
- 如果为满,生产者必须等待,调用condition实例的wait()。
- 消费者可以运行。消费者消耗队列,并产生一个空余位置。
- 然后消费者notify生产者。
- 当消费者释放lock,消费者可以acquire这个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
|
from
threading
import
Thread
,
Condition
import
time
import
random
queue
=
[
]
MAX_NUM
=
10
condition
=
Condition
(
)
class
ProducerThread
(
Thread
)
:
def
run
(
self
)
:
nums
=
range
(
5
)
global
queue
while
True
:
condition
.
acquire
(
)
if
len
(
queue
)
==
MAX_NUM
:
print
"Queue full, producer is waiting"
condition
.
wait
(
)
print
"Space in queue, Consumer notified the producer"
num
=
random
.
choice
(
nums
)
queue
.
append
(
num
)
print
"Produced"
,
num
condition
.
notify
(
)
condition
.
release
(
)
time
.
sleep
(
random
.
random
(
)
)
class
ConsumerThread
(
Thread
)
:
def
run
(
self
)
:
global
queue
while
True
:
condition
.
acquire
(
)
if
not
queue
:
print
"Nothing in queue, consumer is waiting"
condition
.
wait
(
)
print
"Producer added something to queue and notified the consumer"
num
=
queue
.
pop
(
0
)
print
"Consumed"
,
num
condition
.
notify
(
)
condition
.
release
(
)
time
.
sleep
(
random
.
random
(
)
)
ProducerThread
(
)
.
start
(
)
ConsumerThread
(
)
.
start
(
)
|
样例输出:
1
2
3
4
5
6
7
8
9
10
11
12
13
|
Produced
0
Consumed
0
Produced
0
Produced
4
Consumed
0
Consumed
4
Nothing
in
queue
,
consumer
is
waiting
Produced
4
Producer
added
something
to
queue
and
notified
the
consumer
Consumed
4
Produced
3
Produced
2
Consumed
3
|
更新:
很多网友建议我在lock和condition下使用Queue来代替使用list。我同意这种做法,但我的目的是展示Condition,wait()和notify()如何工作,所以使用了list。
以下用Queue来更新一下代码。
Queue封装了Condition的行为,如wait(),notify(),acquire()。
现在不失为一个好机会读一下Queue的文档(http://docs.python.org/2/library/queue.html)。
更新程序:
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
|
from
threading
import
Thread
import
time
import
random
from
Queue
import
Queue
queue
=
Queue
(
10
)
class
ProducerThread
(
Thread
)
:
def
run
(
self
)
:
nums
=
range
(
5
)
global
queue
while
True
:
num
=
random
.
choice
(
nums
)
queue
.
put
(
num
)
print
"Produced"
,
num
time
.
sleep
(
random
.
random
(
)
)
class
ConsumerThread
(
Thread
)
:
def
run
(
self
)
:
global
queue
while
True
:
num
=
queue
.
get
(
)
queue
.
task_done
(
)
print
"Consumed"
,
num
time
.
sleep
(
random
.
random
(
)
)
ProducerThread
(
)
.
start
(
)
ConsumerThread
(
)
.
start
(
)
|
解释:
- 在原来使用list的位置,改为使用Queue实例(下称队列)。
- 这个队列有一个condition,它有自己的lock。如果你使用Queue,你不需要为condition和lock而烦恼。
- 生产者调用队列的put方法来插入数据。
- put()在插入数据前有一个获取lock的逻辑。
- 同时,put()也会检查队列是否已满。如果已满,它会在内部调用wait(),生产者开始等待。
- 消费者使用get方法。
- get()从队列中移出数据前会获取lock。
- get()会检查队列是否为空,如果为空,消费者进入等待状态。
- get()和put()都有适当的notify()。现在就去看Queue的源码吧。