生产者消费者模式

在实际的软件开发过程中，经常会碰到如下场景：某个模块负责产生数据，这些数据由另一个模块来负责处理（此处的模块是广义的，可以是类、函数、线程、进程等）。产生数据的模块，就形象地称为生产者；而处理数据的模块，就称为消费者。

单单抽象出生产者和消费者，还够不上是生产者消费者模式。该模式还需要有一个缓冲区处于生产者和消费者之间，作为一个中介。生产者把数据放入缓冲区，而消费者从缓冲区取出数据，如下图所示：

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。生产者和消费者彼此之间不直接通讯，而通过消息队列（缓冲区）来进行通讯，所以生产者生产完数据之后不用等待消费者处理，直接扔给消息队列，消费者不找生产者要数据，而是直接从消息队列里取，消息队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。这个消息队列就是用来给生产者和消费者解耦的。————->这里又有一个问题，什么叫做解耦？

解耦：假设生产者和消费者分别是两个类。如果让生产者直接调用消费者的某个方法，那么生产者对于消费者就会产生依赖（也就是耦合）。将来如果消费者的代码发生变化，可能会影响到生产者。而如果两者都依赖于某个缓冲区，两者之间不直接依赖，耦合也就相应降低了。生产者直接调用消费者的某个方法，还有另一个弊端。由于函数调用是同步的（或者叫阻塞的），在消费者的方法没有返回之前，生产者只好一直等在那边。万一消费者处理数据很慢，生产者就会白白糟蹋大好时光。缓冲区还有另一个好处。如果制造数据的速度时快时慢，缓冲区的好处就体现出来了。当数据制造快的时候，消费者来不及处理，未处理的数据可以暂时存在缓冲区中。等生产者的制造速度慢下来，消费者再慢慢处理掉。

因为太抽象，看过网上的说明之后，通过我的理解，我举了个例子：吃包子。

假如你非常喜欢吃包子（吃起来根本停不下来），今天，你妈妈（生产者）在蒸包子，厨房有张桌子（缓冲区），你妈妈将蒸熟的包子盛在盘子（消息）里，然后放到桌子上，你正在看巴西奥运会，看到蒸熟的包子放在厨房桌子上的盘子里，你就把盘子取走，一边吃包子一边看奥运。在这个过程中，你和你妈妈使用同一个桌子放置盘子和取走盘子，这里桌子就是一个共享对象。生产者添加食物，消费者取走食物。桌子的好处是，你妈妈不用直接把盘子给你，只是负责把包子装在盘子里放到桌子上，如果桌子满了，就不再放了，等待。而且生产者还有其他事情要做，消费者吃包子比较慢，生产者不能一直等消费者吃完包子把盘子放回去再去生产，因为吃包子的人有很多，如果这期间你好朋友来了，和你一起吃包子，生产者不用关注是哪个消费者去桌子上拿盘子，而消费者只去关注桌子上有没有放盘子，如果有，就端过来吃盘子中的包子，没有的话就等待。对应关系如下图：

考察了一下，原来当初设计这个模式，主要就是用来处理并发问题的，而Celery就是一个用python写的并行分布式框架。

然后我接着去学习Celery

Celery 是一个强大的 分布式任务队列 的 异步处理框架，它可以让任务的执行完全脱离主程序，甚至可以被分配到其他主机上运行。我们通常使用它来实现异步任务（async task）和定时任务（crontab）。我们需要一个消息队列来下发我们的任务。首先要有一个消息中间件，此处选择rabbitmq (也可选择 redis 或 Amazon Simple Queue Service(SQS)消息队列服务)。推荐 选择 rabbitmq 。使用RabbitMQ是官方特别推荐的方式，因此我也使用它作为我们的broker。它的架构组成如下图：

通过celery worker 启动的是启动消费者
通过celery beat 是启动任务生产者
python中使用delay生产任务

Celery的定义

Celery（芹菜）是一个简单、灵活且可靠的，处理大量消息的分布式系统，并且提供维护这样一个系统的必需工具。

我比较喜欢的一点是：Celery支持使用任务队列的方式在分布的机器、进程、线程上执行任务调度。然后我接着去理解什么是任务队列。

任务队列

任务队列是一种在线程或机器间分发任务的机制。

消息队列

消息队列的输入是工作的一个单元，称为任务，独立的职程（Worker）进程持续监视队列中是否有需要处理的新任务。

Celery 用消息通信，通常使用中间人（Broker）在客户端和职程间斡旋。这个过程从客户端向队列添加消息开始，之后中间人把消息派送给职程，职程对消息进行处理。如下图所示：

Celery 系统可包含多个职程和中间人，以此获得高可用性和横向扩展能力。

Celery****的架构

Celery的架构由三部分组成，消息中间件（message broker），任务执行单元（worker）和任务执行结果存储（task result store）组成。

消息中间件

Celery本身不提供消息服务，但是可以方便的和第三方提供的消息中间件集成，包括，RabbitMQ,Redis,MongoDB等，这里我先去了解RabbitMQ,Redis。

linux安装redis参考：https://blog.csdn.net/luanpeng825485697/article/details/81205424

docker 安装redis参考：https://blog.csdn.net/luanpeng825485697/article/details/81209596

docker安装rabbitmq参考：https://blog.csdn.net/luanpeng825485697/article/details/82078416

任务执行单元

Worker是Celery提供的任务执行的单元，worker并发的运行在分布式的系统节点中

任务结果存储

Task result store用来存储Worker执行的任务的结果，Celery支持以不同方式存储任务的结果，包括Redis，MongoDB，Django ORM，AMQP等，这里我先不去看它是如何存储的，就先选用Redis来存储任务执行结果。

然后我接着去安装Celery，在安装Celery之前，我已经在自己虚拟机上安装好了Python，版本是3.6，

安装celery，版本为4.2.1

sudo apt install python-celery-common

因为涉及到消息中间件，所以我先去选择一个在我工作中要用到的消息中间件（在Celery帮助文档中称呼为中间人<broker>），为了更好的去理解文档中的例子，我安装了两个中间件，一个是RabbitMQ,一个redis。

在这里我就先根据Celery的帮助文档安装和设置RabbitMQ。要使用 Celery，我们需要创建一个 RabbitMQ 用户、一个虚拟主机，并且允许这个用户访问这个虚拟主机。下面是我在个人pc机Ubuntu16.04上的设置：

$ sudo rabbitmqctl add_user forward password

#创建了一个RabbitMQ用户,用户名为forward，密码是password

$ sudo rabbitmqctl add_vhost ubuntu

#创建了一个虚拟主机，主机名为ubuntu

$ sudo rabbitmqctl set_permissions -p ubuntu forward ".*" ".*" ".*"

#允许用户forward访问虚拟主机ubuntu，因为RabbitMQ通过主机名来与节点通信

$ sudo rabbitmq-server

之后我启用RabbitMQ服务器，结果如下，成功运行：

之后我安装Redis,它的安装比较简单,如下：

$ sudo pip install redis

然后进行简单的配置，只需要设置 Redis 数据库的位置:

BROKER_URL = 'redis://localhost:6379/0'

URL的格式为：

redis://:password**@hostname**:port/db_number

URL Scheme 后的所有字段都是可选的，并且默认为 localhost 的 6379 端口，使用数据库 0。我的配置是：

redis://:password**@ubuntu**:6379/5

之后安装Celery，我是用标准的Python工具pip安装的，如下：

$ sudo pip install celery

celery 的命令

Global Options:
  -A APP, --app APP
  -b BROKER, --broker BROKER
  --result-backend RESULT_BACKEND
  --loader LOADER
  --config CONFIG
  --workdir WORKDIR
  --no-color, -C
  --quiet, -q

---- -- - - ---- Commands- -------------- --- ------------

+ Main: 
|    celery worker     
|    celery events
|    celery beat
|    celery shell
|    celery multi
|    celery amqp

开始使用 Celery

使用celery包含三个方面：1. 定义任务函数。2. 运行celery服务。3. 客户应用程序的调用。

注意：发送者和消费者职能处理启动前就定义好的任务，不能代码中现场定义。也就是说必须在定义任务后才能启动消费者。

定义

为了测试Celery能否工作，我运行了一个最简单的任务，编写app1.py：

from celery import Celery
# broker设置中间件，backend设置后端存储
app = Celery('app_name',broker='redis://127.0.0.1:6379/5',backend='redis://127.0.0.1:6379/6')

@app.task(name='task_name')       # 被标记的都属于这个app的任务
def add(x,y):
    return x+y

编辑保存退出后，我在当前目录下运行如下命令(记得要先开启redis)：

启动一个worker

$ celery  worker -A app1 --loglevel=info

其中celery会根据--app后面的参数作为命令启动目录的相对目录，去找个这个相对目录文件中的app变量，并把所有用@app.task修饰的函数作为任务来记录元数据。所以不同目录的启动命令是不一样的。

#查询文档，了解到该命令中-A参数表示的是Celery APP的名称，这个实例中指的就是app1.py（和文件名一致），后面的app1就是APP的名称，worker是一个执行任务角色，后面的loglevel=info记录日志类型默认是info,这个命令启动了一个worker,用来执行程序中add这个加法任务（task）。

然后看到界面显示结果如下：

 -------------- celery@luanpeng-XPS15R v4.2.1 (windowlicker)
---- **** ----- 
--- * ***  * -- Linux-4.15.0-33-generic-x86_64-with-Ubuntu-16.04-xenial 2018-09-24 19:19:11
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         tasks:0x7f34ba22dcc0
- ** ---------- .> transport:   redis://127.0.0.1:6379/5
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 8 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . tasks.add

[2018-09-24 19:19:11,186: INFO/MainProcess] Connected to redis://127.0.0.1:6379/5
[2018-09-24 19:19:11,193: INFO/MainProcess] mingle: searching for neighbors
[2018-09-24 19:19:12,212: INFO/MainProcess] mingle: all alone
[2018-09-24 19:19:12,224: INFO/MainProcess] celery@luanpeng-XPS15R ready.

我们可以看到Celery正常工作在名称luanpeng-XPS15R的虚拟主机上，版本为v4.2.1，在下面的[config]中我们可以看到当前APP的名称tasks，运输工具transport就是我们在程序中设置的中间人redis://127.0.0.1:6379/5，result我们没有设置，暂时显示为disabled,然后我们也可以看到worker缺省使用perfork来执行并发，当前并发数显示为1，然后可以看到下面的[queues]就是我们说的队列，当前默认的队列是celery,然后我们看到下面的[tasks]中有一个任务task_name.

生产者

生产者发送消息启动。按照生产者所在目录引入app中的task函数（也会把task的名称获取到）。调用delay函数（apply_async()方法的升级版）时会发起这样一个name的task就可以了。

例如在上层目录发起任务

from app1.app1 import add
result = add.delay(1,2)   # apply_async()方法的升级版
print(result)

 

• 1
• 2
• 3

启动后，消费者开始处理消息

[2018-09-24 20:07:11,496: INFO/MainProcess] Received task: app1.tasks.add[8207c280-0864-4b1e-8792-155de5417406]  
[2018-09-24 20:07:11,501: INFO/ForkPoolWorker-4] Task app1.tasks.add[8207c280-0864-4b1e-8792-155de5417406] succeeded in 0.003930353002942866s: 12

 

• 1
• 2

第一行说明worker收到了一个任务：app1.tasks.add,这里我们和之前发送任务返回的AsyncResult对比我们发现，每个task都有一个唯一的ID，第二行说明了这个任务执行succeed,执行结果为12。

查看资料说调用任务后会返回一个AsyncResult实例，可用于检查任务的状态，等待任务完成或获取返回值（如果任务失败，则为异常和回溯）。但这个功能默认是不开启的，需要设置一个 Celery 的结果后端（backend），也就是tasks.py设置的使用redis进行结果存储。

通过这个例子后我对Celery有了初步的了解，然后我在这个例子的基础上去进一步的学习。

因为Celery是用Python编写的，所以为了让代码结构化一些，就像一个应用

app1/app1_app.py

from celery import Celery
import os,io
# 在app1目录同级目录执行celery -A app1.app1_app worker -l info
app = Celery(main='app1.app1_app',include=['app1.tasks1','app1.tasks2'])  # 创建app，并引入任务定义。main、include参数的值为模块名，所以都是指定命令的相对目录

app.config_from_object('app1.app1_config')   # 通过配置文件进行配置，而着这里是相对目录

# broker设置中间件，backend设置后端存储
# app = Celery('app1.app1_app',broker='redis://127.0.0.1:6379/5',backend='redis://127.0.0.1:6379/6',include=['app1.tasks1','app1.task2'])

if __name__ == "__main__":
    log_path = os.getcwd()+'/log/celery.log'
    if(not os.path.exists(log_path)):
        f = open(log_path, 'w')
        f.close()
    # 在app1目录同级目录执行celery -A app1.app1_app worker -l info
    app = Celery(main='app1_app',include=['tasks1', 'tasks2'])  # 创建app，并引入任务定义。main、include参数的值为模块名，所以都是指定命令的相对目录

    app.config_from_object('app1_config')  # 通过配置文件c进行配置，而着这里是相对目录
    # 使用下面的命令也可以启动celery，不过要该模块的名称，是的相对目录正确
    app.start(argv=['celery', 'worker', '-l', 'info', '-f', 'log/celery.log', "-c", "40"])

 

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21
• 22

一定要注意模块的相对目录，和你想要执行命令的目录

#首先创建了一个celery实例app,实例化的过程中，制定了任务名(也就是包名.模块名)，Celery的第一个参数是当前模块的名称，我们可以调用config_from_object()来让Celery实例加载配置模块，我的例子中的配置文件起名为app1_config.py,配置文件如下：

BROKER_URL = 'redis://127.0.0.1:6379/5'   # 配置broker  中间件
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/6'   # 配置backend结果存储
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

 

• 1
• 2
• 3
• 4
• 5

在配置文件中我们可以对任务的执行等进行管理，比如说我们可能有很多的任务,但是我希望有些优先级比较高的任务先被执行,而不希望先进先出的等待。那么需要引入一个队列的问题. 也就是说在我的broker的消息存储里面有一些队列，他们并行运行，但是worker只从对应 的队列里面取任务。在这里我们希望tasks.py中的某些任务先被执行。task中我设置了两个任务：

所以我通过from celery import group引入group,用来创建并行执行的一组任务。然后这块现需要理解的就是这个@app.task,@符号在python中用作函数修饰符，到这块我又回头去看python的装饰器(在代码运行期间动态增加功能的方式)到底是如何实现的，在这里的作用就是通过task()装饰器在可调用的对象（app）上创建一个任务。

tasks1.py

from app1.app1_app import app

@app.task
def deal1(text):
    print(text)
    return text+"======="

 

• 1
• 2
• 3
• 4
• 5
• 6

tasks2.py

from app1.app1_app import app

@app.task
def deal2(text):
    print(text)
    return text+"+++++++++"

 

• 1
• 2
• 3
• 4
• 5
• 6
• 7

队列

了解完装饰器后，我回过头去整理配置的问题，前面提到任务的优先级问题，在这个例子中如果我们想让deal1这个任务优先于deal2任务被执行，我们可以将两个任务放到不同的队列中，由我们决定先执行哪个任务，我们可以在配置文件app1_config.py中增加这样配置：

URL的格式为：

redis://:password@hostname:port/db_number

 

• 1

from kombu import Exchange,Queue

BROKER_URL = 'redis://127.0.0.1:6379/5'   # 配置broker  中间件
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/6'   # 配置backend结果存储
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# (当使用Redis作为broker时，Exchange的名字必须和Queue的名字一样)
CELERY_QUEUES = (
                    Queue("default", Exchange("default"), routing_key = "default"),
                    Queue("for_task1", Exchange("for_task1"), routing_key="task_a"),
                    Queue("for_task2", Exchange("for_task2"), routing_key="task_b")
)

# 定义任务的走向，不同的任务发送  进入不同的队列，并为不同的任务设定不同的routing_key
# 若没有指定这个任务route到那个Queue中去执行，此时执行此任务的时候，会route到Celery默认的名字叫做celery的队列中去。
CELERY_ROUTES = {
    'app1.tasks1.deal1': {"queue": "for_task1", "routing_key": "task_a"},
    'app1.tasks2.deal2':{"queue":"for_task2","routing_key":"task_b"}
}

 

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20
• 21

先了解了几个常用的参数的含义：

Exchange:交换机，决定了消息路由规则；

Queue:消息队列；

Channel:进行消息读写的通道；

Bind:绑定了Queue和Exchange，意即为符合什么样路由规则的消息，将会放置入哪一个消息队列；

我将deal1这个函数任务放在了一个叫做for_task1的队列里面，将deal2这个函数任务放在了一个叫做for_task2的队列里面，然后我在当前应用目录下执行命令：

celery -A app1.app1_app worker -l info -Q for_task1

 

• 1

这个worker就只负责处理for_task1这个队列的任务，这是通过在启动worker是使用-Q Queue_Name参数指定的。

我们定义任务调用文件start_task.py

from __future__ import print_function
from app1.app1_app import app


if __name__=="__main__":
    for i in range(10):
        text = 'text'+str(i)
        app.send_task('app1.tasks1.deal1',args=[text])   # 任务的名称必须和Celery注册的任务名称相同
        app.send_task('app1.tasks2.deal2',args=[text])  # 任务的名称必须和Celery注册的任务名称相同
        print('push over %d'%i)

 

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12

执行上述代码文件

python start_task.py

 

• 1

任务已经被执行，我在worker控制台查看结果（只有app1.appa_app.deal1任务被这个worker执行了）：

[2018-09-24 22:26:38,928: INFO/ForkPoolWorker-8] Task app1.tasks1.deal1[b3007993-9bfb-4161-b5b2-4f0f022f2f8b] succeeded in 0.0008255800021288451s: 'text4======='
[2018-09-24 22:26:38,928: INFO/ForkPoolWorker-6] Task app1.tasks1.deal1[df24b991-88fc-4253-86bf-540754c62da9] succeeded in 0.004320767002354842s: 'text3======='
[2018-09-24 22:26:38,929: INFO/MainProcess] Received task: app1.tasks1.deal1[dbdf9ac0-ea27-4455-90d2-e4fe8f3e895e]  
[2018-09-24 22:26:38,930: WARNING/ForkPoolWorker-4] text5
[2018-09-24 22:26:38,931: INFO/ForkPoolWorker-4] Task app1.tasks1.deal1[dbdf9ac0-ea27-4455-90d2-e4fe8f3e895e] succeeded in 0.0006721289973938838s: 'text5======='

 

• 1
• 2
• 3
• 4
• 5
• 6

可以看到worker收到任务，并且执行了任务。

Scheduler ( 定时任务，周期性任务 )

在这里我们还是在交互模式下手动去执行，我们想要crontab的定时生成和执行，我们可以用celery的beat去周期的生成任务和执行任务，在这个例子中我希望每10秒钟产生一个任务，然后去执行这个任务，我可以这样配置（在app1_config.py文件中添加如下内容）：

# 设计周期任务
CELERY_TIMEZONE = 'Asia/Shanghai'

from celery.schedules import crontab   # 设置定时任务
from datetime import timedelta
# 每隔30秒执行app1.tasks1.deal函数
CELERYBEAT_SCHEDULE = {
    'deal-every-30-seconds': {
         'task': 'app1.tasks1.deal1',
         'schedule': timedelta(seconds=30),
         'args': ['hello']
    },
    'deal-every-10-seconds': {
         'task': 'app1.tasks2.deal2',
         'schedule': timedelta(seconds=10),
         'args': ['hello']
    },
     # Executes every Monday morning at 7:30 A.M
    'deal-every-monday-morning': {
         'task': 'app1.tasks2.deal2',
         'schedule': crontab(hour=7, minute=30, day_of_week=1),
         'args': ['hello']
    },
}

 

• 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

使用了scheduler，要制定时区：CELERY_TIMEZONE = ‘Asia/Shanghai’，启动celery加上-B的参数。

celery -A app1.app1_app worker -l info -B

 

• 1

前两个任务为周期任务，第三个任务为定时任务，指定时间点开始执行分发任务，让worker取走执行，可以这样配置：

看完这些基础的东西，我回过头对celery在回顾了一下，用图把它的框架大致画出来，如下图：

celery任务调度的配置项

https://docs.celeryproject.org/en/latest/userguide/configuration.html

celery 任务监控

celery flower