deque实现生产者-消费者队列

1.概述

使用python内置的list类型可以实现一个生产-消费队列功能，这个队列是先进先出。把批量的数据放到生产队列可以加快程序处理业务的速度，然后消费者可以从消费队列获取数据在后台慢慢处理数据。

2.普通方式实现实现生产者-消费者队列性能

2.1.使用list实现生产者-消费者队列代码

下面代码实现了一个处理Email的功能，首先接收Email存放到生产队列，消费者从队列中获取一个Email对象进行处理。
代码的实现分为三个部分，一个是创建一个Email对象，然后是创建生产队里和消费队列，最后调用该业务实现邮件的生产和消费功能。

'''创建Email对象'''
# Example 1
class Email:def __init__(self, sender, receiver, message):self.sender = senderself.receiver = receiverself.message = message'''发送Email'''
# Example 2
def get_emails():yield Email('foo@example.com', 'bar@example.com', 'hello1')yield Email('baz@example.com', 'banana@example.com', 'hello2')yield Noneyield Email('meep@example.com', 'butter@example.com', 'hello3')yield Email('stuff@example.com', 'avocado@example.com', 'hello4')yield Noneyield Email('thingy@example.com', 'orange@example.com', 'hello5')yield Email('roger@example.com', 'bob@example.com', 'hello6')yield Noneyield Email('peanut@example.com', 'alice@example.com', 'hello7')yield NoneEMAIL_IT = get_emails()class NoEmailError(Exception):passdef try_receive_email():# Returns an Email instance or raises NoEmailErrortry:email = next(EMAIL_IT)except StopIteration:email = Noneif not email:raise NoEmailErrorprint(f'Produced email: {email.message}')return email'''生产队列，调用try_receive_email函数接收发送的Emial邮件，放到生产队列'''
# Example 3
def produce_emails(queue):while True:try:email = try_receive_email()except NoEmailError:returnelse:queue.append(email)  # Producer'''消费队列'''
# Example 4
def consume_one_email(queue):if not queue:returnemail = queue.pop(0)  # Consumer# Index the message for long-term archivalprint(f'Consumed email: {email.message}')'''编排业务逻辑'''
# Example 5
def loop(queue, keep_running):# 每次调用make_test_end里面的闭包函数func，判断是否循环。while keep_running():produce_emails(queue)consume_one_email(queue)def make_test_end():count=list(range(10))def func():if count:count.pop()return Truereturn Falsereturn funcdef my_end_func():passmy_end_func = make_test_end()
# 传入队列和循环次数
loop([], my_end_func)

2.2.timeit测试性能

下面我们对生产队列和消费队列分别进行性能测试，这次测试使用timeit模块提供的函数测试执行耗时。

1.timeit模块使用介绍

timeit模块下主要有两个函数十分有用，分别为timeit.timeit、timeit.repeat
timeit函数和repeat函数使用上非常相似，repeat函数之比timeit多了一个repeat参数

timeit函数参数介绍

setup：这个参数可以将stmt的环境传进去。比如各种import以及参数。多个值用分号；分隔开
stmt：指定要执行的语句/statement,值可以是字符串形式的表达式，也可以是一个函数，或者是一个变量的形式。
number：指定stmt语句执行的次数，默认值为一百万次

repeat函数介绍：timeit函数参数它都有，只多了一个repeat参数

repeat：指定重复的次数，该参数是指定number参数的重复次数。比如number=3, repeat=2那么就会重复两次，每次执行3次一共执行6次。

setup参数只看上面的介绍可能不太理解它的使用，下面使用示例介绍它的用法。

from timeit import timeit
# stmt是执行的内容，setup为当前执行的内容导入依赖环境和参数
res=timeit(stmt="json.loads(json_data)",number=1000,setup="import json;data={'name':'egon','age':18};json_data=json.dumps(data)")
print(res)

2.测试生产队列性能

生产队列的核心代码是向列表添加数据，下面只对该代码做测试。

'''测试生产队列性能'''
# Example 6
from timeit import timeit, repeat# 输出测试结果
def print_results(count, tests):avg_iteration = sum(tests) / len(tests)print(f'Count {count:>5,} takes {avg_iteration:.6f}s')return count, avg_iteration# 测试执行时间
def list_append_benchmark(count):def run(queue):for i in range(count):queue.append(i)tests = repeat(setup='queue = []',stmt='run(queue)',globals=locals(),repeat=1000,number=1)return print_results(count, tests)# Example 7
# 统计不同的执行次数对测试结果的影响
def print_delta(before, after):before_count, before_time = beforeafter_count, after_time = aftergrowth = 1 + (after_count - before_count) / before_countslowdown = 1 + (after_time - before_time) / before_timeprint(f'{growth:>4.1f}x data size, {slowdown:>4.1f}x time')# 调用测试函数，测试性能。首先测试500次然后分别测试500的倍数输出的结果
baseline = list_append_benchmark(500)
for count in (1_000, 2_000, 3_000, 4_000, 5_000):print()comparison = list_append_benchmark(count)print_delta(baseline, comparison)

运行上面的代码，分别输出了对应次数执行的时间。随着数据量变大，通过append把数据假如到列表的耗时也成比例的增加。

Count   500 takes 0.000037sCount 1,000 takes 0.000072s2.0x data size,  2.0x timeCount 2,000 takes 0.000145s4.0x data size,  3.9x timeCount 3,000 takes 0.000219s6.0x data size,  5.9x timeCount 4,000 takes 0.000278s8.0x data size,  7.5x timeCount 5,000 takes 0.000349s
10.0x data size,  9.4x time

3.测试消费队列

'''测试消费队列'''
# Example 8
def list_pop_benchmark(count):def prepare():return list(range(count))def run(queue):while queue:queue.pop(0)tests = repeat(setup='queue = prepare()',stmt='run(queue)',globals=locals(),repeat=1000,number=1)return print_results(count, tests)# Example 9
baseline = list_pop_benchmark(500)
for count in (1_000, 2_000, 3_000, 4_000, 5_000):print()comparison = list_pop_benchmark(count)print_delta(baseline, comparison)

运行上面的代码，可以发现用pop(0)从队列开头移除元素所花费的时间跟队列的长度呈平方关系。这是因为列表底层是用数组结构存储数据，用pop(0)删除开头的元素，需要把后面的元素都要向前移动一个位置，相当于把列表内容都修改了一遍。

Count   500 takes 0.000058sCount 1,000 takes 0.000142s2.0x data size,  2.5x timeCount 2,000 takes 0.000395s4.0x data size,  6.9x timeCount 3,000 takes 0.000767s6.0x data size, 13.3x timeCount 4,000 takes 0.001331s8.0x data size, 23.1x timeCount 5,000 takes 0.002514s
10.0x data size, 43.6x time

3.使用deque实现生产者-消费者队列

python内置的collections模块里有个deque类可以解决这个问题，这个类实现的是双向队列，从头部插入或尾部删除数据都只需要固定时间，非常适合充当FIFO队列。

3.1.测试生产队列

def deque_append_benchmark(count):def prepare():return collections.deque()def run(queue):for i in range(count):queue.append(i)tests = repeat(setup='queue = prepare()',stmt='run(queue)',globals=locals(),repeat=1000,number=1)return print_results(count, tests)baseline = deque_append_benchmark(500)
for count in (1_000, 2_000, 3_000, 4_000, 5_000):print()comparison = deque_append_benchmark(count)print_delta(baseline, comparison)

运行上面测试代码，可以看到它的append方法在队列边长之后，所花费的时间通list队列近似相似，1：1增长。

Count   500 takes 0.000032sCount 1,000 takes 0.000066s2.0x data size,  2.1x timeCount 2,000 takes 0.000137s4.0x data size,  4.3x timeCount 3,000 takes 0.000204s6.0x data size,  6.4x timeCount 4,000 takes 0.000271s8.0x data size,  8.5x timeCount 5,000 takes 0.000340s
10.0x data size, 10.6x time

3.2.测试消费队列

# Example 12
def dequeue_popleft_benchmark(count):def prepare():return collections.deque(range(count))def run(queue):while queue:queue.popleft()tests = repeat(setup='queue = prepare()',stmt='run(queue)',globals=locals(),repeat=1000,number=1)return print_results(count, tests)baseline = dequeue_popleft_benchmark(500)
for count in (1_000, 2_000, 3_000, 4_000, 5_000):print()comparison = dequeue_popleft_benchmark(count)print_delta(baseline, comparison)

运行上面测试代码，消费队列所花费的时间与append方法近似，也随着队列长度线性增长不会早先呈平方增长。

Count   500 takes 0.000027sCount 1,000 takes 0.000056s2.0x data size,  2.1x timeCount 2,000 takes 0.000116s4.0x data size,  4.3x timeCount 3,000 takes 0.000177s6.0x data size,  6.5x timeCount 4,000 takes 0.000233s8.0x data size,  8.6x timeCount 5,000 takes 0.000294s
10.0x data size, 10.9x time