消息队列是什么

很多人初次听说消息队列的时候可能会觉得这个词有点高级，一定充满了复杂的知识点，对，其实没错，生产中使用使的确很复杂，但在学习时，我们可以将其理解的很简单，怎么理解呢，拆开来看，消息（Message）+队列（Queue）：

• 消息就很好理解了，微信消息，短信消息，小道消息，内幕消息等等，消息在日常生活中简直无所不在，这里的消息也并无差别，不过我还是简单用“黑话”描述一下，一段承载生产实体传递到消费实体通信内容的结构化数据，通常是序列化的字节数组。
• 队列那就更简单了，数据结构学过吧，先进先出知道吧，没了。
  消息队列其实是生产者-消费者模型的一种实现方式，如下图所示，把内存缓冲区换成消息队列即可很清晰的表达其作用：作为生产者至消费者之间的一个消息通道。
  
  image.png
  
  图片来源

为什么要用消息队列

这里我用一段文字做一个形象的比喻：
《若》
一阵凉风吹过树林，几片枯黄的叶子在树枝上摇摇欲坠，摩擦中还发出沙沙之声，仿佛都想把对方先推下去，殊不知无论早晚，结局都是一样。
风继续向前吹去，却撞在了一座宏大的建筑上，不敢再向前。前面的建筑像是一座住着吸血鬼的中世纪城堡，墙壁上雕刻着奇异的图案，最上方的房檐上站立着几只黑色的石鹰，城堡两侧隐藏在浓密的迷雾中让人看不太清，但很明显一定充满了极致的对称美。
此时城堡的大厅内零零散散站着一些人，但却异常安静，安静到空气中充满了肃杀之气，配合着这秋意的萧瑟使人不寒而栗。
这些人有同一个身份--赏金猎人，他们来此只有一个目的，那块悬浮在大厅上空的木牌。木牌看似没什么特别之处，但边角处暗红色的血迹和剑痕几乎在明示其并不简单。
木牌有一个很文雅的名字，“见无”，意思一目了然，在其上被见到的人马上就会在这个世界消失。
“见无”上的任务多是由一些富商巨贾，豪门贵族发布，一是他们掌握敌国的财富，请得动这些赏金猎人，二是总还得维持在江湖上那点伪善。虽说“见无”，但事情总有例外，高居任务榜首的天级、地级、玄级三个任务已在那里七百年之久，而今天让这些神龙见首不见尾的猎人们齐聚一厅，抬首观望的原因就是，玄级任务被完成了。
玄级任务不知由谁发布，但任务内容从发布至今却未曾变化：当代姑苏慕容家家主的项上人头。岁月更迭，慕容家主不知换了多少代，却未有一人是死于这个任务，反而因任务丢掉性命的赏金猎人那真是一茬又一茬，甚至慕容家已经将此作为其震慑武林的手段。
关于玄级任务的发布者，坊间猜测颇多，有位于西南把控天下古玩的星家，虎踞辽北镇守边疆的岳家，甚至还有猜测是隐世不出根植燕京的王家，每种猜测都有其缘由，不必细说。
此时人们更关注的是，慕容家主被杀了？那可是慕容家，一门冷月剑法传承了千年，也震慑了武林千年的慕容家，有道是，“冷光浮照千万里，月下再无一丝声“。不过“见无”万千年来从未出过差错，猎人们不存在一丝对结果的质疑，至于为什么没有聚在一起讨论是谁完成了任务，是因为所有人心中都是同一个答案，他叫“若”，没有姓氏没有出身，甚至没人见过他的脸，只知道他的名字是“若”，一个仿若虚无的人。
“若”的强大没有人敢质疑，他曾。。。

在上面的小故事中，任务榜单对应消息队列，任务发布人是生产者，赏金猎人是消费者。
先来想像一下如果没有任务榜单，一个富商想杀死某人只能挨个给猎人们打电话，“喂，你有空不啦，我要杀xxx，你要多少钱”，“没空”。发现问题了吗，这种方式效率是极其低下的并且想给人家打电话还得知道电话号码，而采用任务榜单就很优雅的解决了这个问题，有任务放上去就好，自然会有人处理。
这就是消息队列的第一个好处，实现了逻辑上的解耦，没有依赖，每个实体各司其职，岂不美哉。并且如果任务极多的话，通过榜单可以分发到很多赏金猎人，每个赏金猎人都可以满负荷干活，这就是分布式。
再思考一下，榜单上的任务其实并不是要马上完成，任务完成后猎人再通过榜单通知发布者即可，这就是异步，异步也是计算机世界应用非常广泛的设计。
当然还可以利用消息队列做更多事情，像上面故事中的任务分级等等，还可以设置不同的路由，某个任务只针对某些猎人可见。当任务过多时，只需要再招募猎人，实现了水平扩展。
总结一下，三大好处：解耦，分布式，异步。应用时可扩展出，不同路由策略、优先级、限流、水平扩展等等好处。

AMQP & rabbitMQ

维基百科：高级消息队列协议即Advanced Message Queuing Protocol（AMQP）是面向消息中间件提供的开放的应用层协议，其设计目标是对于消息的排序、路由（包括点对点和订阅-发布）、保持可靠性、保证安全性^[1]。AMQP规范了消息传递方和接收方的行为，以使消息在不同的提供商之间实现互操作性，就像SMTP，HTTP，FTP等协议可以创建交互系统一样。与先前的中间件标准（如Java消息服务）不同的是，JMS在特定的API接口层面和实现行为上进行了统一，而高级消息队列协议则关注于各种消息如何以字节流的形式进行传递。因此，使用了符合协议实现的任意应用程序之间可以保持对消息的创建、传递。
官网：The Advanced Message Queuing Protocol (AMQP) is an open standard for passing business messages between applications or organizations. It connects systems, feeds business processes with the information they need and reliably transmits onward the instructions that achieve their goals.
概念性的描述看看就好，直接理解起来还是有些困难的，从拆分组件的角度看就简单多了。
AMQP中的组件：
Broker：rabbitMQ服务就是Broker，是一个比较大的概念，描述的是整个应用服务。
交换机（exchange）：用于接收来自生产者的消息，并把消息转发到消息队列中。AMQP中存在四种交换机，Direct exchange、Fanout exchange、Topic exchange、Headers exchange，区别
消息队列（message queuq）：上文说了。
binding：描述消息队列和交换机的绑定关系，使用routing key描述。

image.png

本文主角--celery

celery概述

celery是python实现的一个轻量级分布式框架系统，使用celery可以很简单快速的实现任务分布式下发。
celery还提供了极其完善的文档，让开发者可以很快速的上手和深入学习。
celery的应用场景就不说了，看了上面消息队列的介绍应该很清楚，什么场景可以用celery。

broker，backend，worker是什么

broker在上文也有提到，可以在此简单理解为用来传递celery任务消息的中间件。最新的celery5中支持四种broker：RabbitMQ、Redis、Amazon SQS、Zookeeper（实验性质）。一般最常用的就是redis和rabbitMQ。
backend是用来存储任务结果和中间状态的实体，backend的选择就很多了，redis/mongoDB/elsticsearch/rabbitMQ，甚至还可以自己声明。如果生产者或者其他服务需要关心异步任务的结果则一定要配置backend。
worker，顾名思义，其作用就是执行任务，需要注意的是启动worker时一般需要设置其监听的队列和最大并发数。

一个简单的demo

root_dir
├── celery_demo
│   ├── __init__.py

init.py

import time

from celery import Celery
from celery.exceptions import TimeoutError
from celery.result import AsyncResult
from kombu import Queue, Exchange

# celery配置，4.0之后引入了小写配置，这种大写配置在6.0之后将不再支持
# 可以参考此链接
# https://docs.celeryproject.org/en/stable/userguide/configuration.html?highlight=worker#std-setting-enable_utc
CONFIG = {
    # 设置时区
    'CELERY_TIMEZONE': 'Asia/Shanghai',
    # 默认为true，UTC时区
    'CELERY_ENABLE_UTC': False,
    # broker，注意rabbitMQ的VHOST要给你使用的用户加权限
    'BROKER_URL': 'amqp://root:root@192.168.1.5:5672/dev',
    # backend配置，注意指定redis数据库
    'CELERY_RESULT_BACKEND': 'redis://192.168.1.5:30412/4',
    # worker最大并发数
    'CELERYD_CONCURRENCY': 10,
    # 如果不设置，默认是celery队列，此处使用默认的直连交换机，routing_key完全一致才会调度到celery_demo队列
    # 此处注意，元组中只有一个值的话，需要最后加逗号
    'CELERY_QUEUES': (
        Queue("celery_demo", Exchange("celery_demo"), routing_key="celery_demo"),
    )
}
app = Celery()
app.config_from_object(CONFIG)


@app.task(name='demo_task')
def demo_task(x, y):
    print(f"这是一个demo任务，睡了10秒，并返回了{x}+{y}的结果。")
    time.sleep(10)
    return x + y


def call():
    def get_result(task_id):
        res = AsyncResult(task_id)
        try:
            # 拿到异步任务的结果，需要用task_id实例化AsyncResult，再调用get方法，get默认是阻塞方法，提供timeout参数，此处设置为0.1秒
            res.get(0.1)
            return res.get(0.1)
        except TimeoutError:
            return None

    tasks = []
    print("开始下发11个任务")
    for _ in range(11):
        tasks.append(demo_task.apply_async((_, _), routing_key='celery_demo', queue='celery_demo'))
    print("等待10秒后查询结果")
    time.sleep(10)
    for index, task in enumerate(tasks):
        task_result = get_result(task.id)
        if task_result is not None:
            print(f"任务{index}的返回值是：{task_result}")
        else:
            print(f"任务{index}还没执行结束")
    print("再等待10秒")
    time.sleep(10)
    print(f"任务10的返回值是：{get_result(tasks[-1].id)}")


if __name__ == '__main__':
    call()

pycharm中启动worker

pycharm-worker.png

celery_demo_queue.png

启动后可以在rabbitMQ的监控看板上看到出现了celery_demo队列。
因为设置了最大并发为10，接下来下发11个任务看是什么结果。

display.png

常用配置说明

介绍一些常用的配置

• CELERYD_PREFETCH_MULTIPLIER：预取消息数量，默认会取4*并发数，在上面的例子中则会最多预取40个消息，如果设置为1，则表示禁止预取。
• CELERY_ACKS_LATE：默认为FALSE，这个参数的理解需要先了解下rabbitMQ的ACK机制，简单说就是如果设置True则会在任务执行完成后才会对消息队列发送确认，表明消息已被消费，如果任务执行中发生了异常情况，未发送确认消息，则消息队列会继续保留此消息，直到下一个worker取走并成功执行。此时衍生出了一个问题，需要保证此消息是幂等的，也就是无论执行多少次结果都一样，否则可能会得到一些预料之外的结果。
• CELERYD_MAX_TASKS_PER_CHILD：worker执行多少个任务会重启进程，默认为无限制，建议设置此值，可避免内存泄漏。
• CELERY_ROUTES：可以在配置中指定每个任务的路由规则，上面例子使用的动态指定队列的方式，在调用时指定路由规则。

其余配置参考官方文档

信号（Signals）

celery中的信号我理解为就是钩子函数，celery提供了不同类型的钩子函数，分别对应不同组件，常用的有三种：任务类型、worker类型、日志类型。
下面附上代码示例，以演示信号的效果：

@after_task_publish.connect
def task_sent_handler(sender=None, headers=None, body=None, **kwargs):
    # information about task are located in headers for task messages
    # using the task protocol version 2.
    info = headers if 'task' in headers else body
    print('after_task_publish for task id {info[id]}'.format(
        info=info,
    ))

@celeryd_after_setup.connect
def setup_direct_queue(sender, instance, **kwargs):
    queue_name = '{0}.dq'.format(sender)  # sender is the nodename of the worker
    worker_logger.info(f"为worker新增一个监控队列：{queue_name}")
    instance.app.amqp.queues.select_add(queue_name)
    worker_logger.info(f"worker当前监控队列：{','.join(instance.app.amqp.queues.keys())}")
@after_setup_logger.connect
def setup_logger(logger, loglevel, logfile, **kwargs):
    worker_logger.info(f"worker日志级别是：{loglevel}")
    worker_logger.info(
        f"logger中目前有{len(logger.handlers)}个handler，分别是：{','.join(type(_).__name__ for _ in logger.handlers)}")

结果如下所示：

task_signals.png

logger_signals.png

设置任务优先级

celery使用rabbitMQ支持任务优先级非常简单，只需要在Queue的配置中加一个参数，并新增CELERY_ACKS_LATE，CELERYD_PREFETCH_MULTIPLIER配置，作用在上个章节有讲，如下

# 优先级范围设置为0-9，最大设置为255，数字越大优先级越高
'CELERY_ACKS_LATE': True,
'CELERYD_PREFETCH_MULTIPLIER': 1,
'CELERY_QUEUES': (
        Queue("celery_demo", Exchange("celery_demo"), routing_key="celery_demo", queue_arguments={'x-max-priority': 9}),
    )

需要注意的是，x-max-priority是rabbitMQ3.5.0版本后才支持的，不要用错版本哟
首先删除之前的celery_demo队列，再次启动worker后可以发现celery_demo队列多了Pri标识，表明已经支持优先级。

priority_queue.png

@task_received.connect
def on_task_received(request, **kwargs):
    # 函数的一个参数就是任务编号
    worker_logger.info(f"任务{request.args[0]}已被worker接收，开始执行")

再改造一下call方法，因为worker的并发数是10，所以先下发10个任务，让worker满并发，再以优先级由低到高下发10个任务，按照预期任务的执行顺序应该task19到task10排列。

def priority_call():
    tasks = []
    print("同时下发20个任务")
    for _ in range(10):
        # apply_async提供priority参数指定优先级
        tasks.append(demo_task.apply_async((_, _), routing_key='celery_demo', queue='celery_demo', priority=0))
    time.sleep(1)
    for _ in range(10, 20):
        # apply_async提供priority参数指定优先级
        tasks.append(demo_task.apply_async((_, _), routing_key='celery_demo', queue='celery_demo', priority=_ % 10))

结果如下，很明显可以看到，符合预期

priority_result.png

不知道有没有人对这两个配置有疑问，看到网上有一些文章提到了一定要配置这两个参数才能实现优先级，但是并没有说具体原因，简单说下我的理解。上文已经说明这两个配置的作用，不再赘述，只说下和优先级的关系。

'CELERY_ACKS_LATE': True,
'CELERYD_PREFETCH_MULTIPLIER': 1,

先假设如果不设置CELERY_ACKS_LATE，celery为提高性能会在任务真正执行前就会向队列发送确认消息，这会导致尽管一个worker设置了并发数是10，但实际上在此worker上最多同时会有20个任务，其中10个正在运行，另外10个是还没发送确认消息（ACK）的，这10个实际上并未开始运行，所以如果其优先级很高，但是却并未执行，也没有分配到其他worker，反而可能低优先级却在其他worker开始执行了，显然不符合优先级的预期。
CELERYD_PREFETCH_MULTIPLIER的作用也是如此，你可以按照上面的分析自行理解下，其目的都是为了保证每个worker的并发都只会分配一个任务。

工作流任务

官方文档写的非常好，有耐心还是去读文档比较好。
我们先来了解下signature，翻译过来就是签名，这里和java的方法签名概念类似，java中用方法名和参数类型组成了一个方法的签名，celery中的signature同样是包装了task和指定的参数，方便可以可以对其进行传递，比如作为参数传递到某个函数。签名包装后还可以进行二次修改，比如新增或更新某个参数，当然还可以通过immutable=True将其设置为不可修改。

celery通过继承Signature实现了几个易用的工作流任务类：

• chain：链式任务，串行执行，父任务的返回值会作为参数传递给子任务
• group：组任务，并行执行，使用celery.result.GroupResult获取结果
• chord：依赖一个group任务，group任务结束后，将所有子任务的返回值作为参数传递给chord任务
• chunks：一般用于将同一任务的极多次执行分组下发，以降低消息传输的成本

工作流任务通常在一组任务有执行顺序的要求时才会用到，做过DAG任务调度工具的同学肯定会容易理解，我举个小例子说明下：
早上起床后的流程：穿衣服->洗漱->吃早餐，这就是一个串行执行的链式任务，可能吃早餐的时候还会看下新闻，吃早餐和看新闻就是一个并行执行的组任务
四个示例

from celery import chain, group, chord, chunks

@app.task(name='demo_task2')
def demo_task2(x, y):
    return x * y


@app.task(name='tsum')
def tsum(nums):
    return sum(nums)


def chain_call():
    # 1 * 2 * 3 * 4 = 24
    # .s()是.signature(）的缩写
    # 还可通过管道符调用chain，具体参考文档
    res = chain(
        *[demo_task2.signature(_, routing_key='celery_demo', queue='celery_demo') for _ in [(1, 2), (3,), (4,)]])()
    print(res.id)
    print(f"chain任务：1 * 2 * 3 * 4={res.get()}")


def group_call():
    res = group(
        *[demo_task2.signature(_, routing_key='celery_demo', queue='celery_demo') for _ in [(1, 2), (3, 4), (5, 6)]])()
    print(res.id)
    print(f"chain任务：1 * 2, 3 * 4, 5 * 6={res.get()}")


def chord_call():
    res = chord(
        (demo_task2.signature(_, routing_key='celery_demo', queue='celery_demo') for _ in [(1, 2), (3, 4), (5, 6)]),
        tsum.s().set(routing_key='celery_demo', queue='celery_demo')
    )()
    print(res.id)
    print(f"chord任务：sum(1 * 2, 3 * 4, 5 * 6)={res.get()}")


def chunk_call():
    res = chunks(demo_task2.s(), [(1, 2), (3, 4), (5, 6)], 2).apply_async(routing_key='celery_demo', queue='celery_demo')
    print(res.id)
    print(f"chunk任务：1 * 2, 3 * 4={res.get()[0]}, 5 * 6={res.get()[1]}")

全部代码，可直接执行

import time

from celery import Celery
from celery.exceptions import TimeoutError
from celery.result import AsyncResult, GroupResult
from kombu import Queue, Exchange
from celery.signals import after_task_publish, celeryd_after_setup, after_setup_logger, task_received, task_success
from celery.utils.log import get_logger, worker_logger
from celery import chain, group, chord, chunks
import logging

# celery配置，4.0之后引入了小写配置，这种大写配置在6.0之后将不再支持
# 可以参考此链接
# https://docs.celeryproject.org/en/stable/userguide/configuration.html?highlight=worker#std-setting-enable_utc
CONFIG = {
    # 设置时区
    'CELERY_TIMEZONE': 'Asia/Shanghai',
    # 默认为true，UTC时区
    'CELERY_ENABLE_UTC': False,
    # broker，注意rabbitMQ的VHOST要给你使用的用户加权限
    'BROKER_URL': 'amqp://root:root@192.168.1.5:5672/dev',
    # backend配置，注意指定redis数据库
    'CELERY_RESULT_BACKEND': 'redis://192.168.1.5:30412/4',
    # worker最大并发数
    'CELERYD_CONCURRENCY': 10,
    # 如果不设置，默认是celery队列，此处使用默认的直连交换机，routing_key完全一致才会调度到celery_demo队列
    'CELERY_ACKS_LATE': True,
    'CELERYD_PREFETCH_MULTIPLIER': 1,
    # 此处注意，元组中只有一个值的话，需要最后加逗号
    'CELERY_QUEUES': (
        Queue("celery_demo", Exchange("celery_demo"), routing_key="celery_demo", queue_arguments={'x-max-priority': 9}),
    )
}
app = Celery()
app.config_from_object(CONFIG)


@app.task(name='demo_task')
def demo_task(x, y):
    time.sleep(10)
    return x + y


@app.task(name='demo_task2')
def demo_task2(x, y):
    return x * y


@app.task(name='tsum')
def tsum(nums):
    return sum(nums)


@celeryd_after_setup.connect
def setup_direct_queue(sender, instance, **kwargs):
    queue_name = '{0}.dq'.format(sender)  # sender is the nodename of the worker
    worker_logger.info(f"为worker新增一个监控队列：{queue_name}")
    instance.app.amqp.queues.select_add(queue_name)
    worker_logger.info(f"worker当前监控队列：{','.join(instance.app.amqp.queues.keys())}")


@after_task_publish.connect
def task_sent_handler(sender=None, headers=None, body=None, **kwargs):
    # information about task are located in headers for task messages
    # using the task protocol version 2.
    info = headers if 'task' in headers else body
    print('after_task_publish for task id {info[id]}'.format(
        info=info,
    ))


@task_received.connect
def on_task_received(request, **kwargs):
    worker_logger.info(f"任务{request.args[0]}已被worker接收，开始执行")


@after_setup_logger.connect
def setup_logger(logger, loglevel, logfile, **kwargs):
    worker_logger.info(f"worker日志级别是：{loglevel}")
    worker_logger.info(
        f"logger中目前有{len(logger.handlers)}个handler，分别是：{','.join(type(_).__name__ for _ in logger.handlers)}")


def call():
    def get_result(task_id):
        res = AsyncResult(task_id)
        try:
            # 拿到异步任务的结果，需要用task_id实例化AsyncResult，再调用get方法，get默认是阻塞方法，提供timeout参数，此处设置为0.1秒
            res.get(0.1)
            return res.get(0.1)
        except TimeoutError:
            return None

    tasks = []
    print("开始下发11个任务")
    for _ in range(11):
        tasks.append(demo_task.apply_async((_, _), routing_key='celery_demo', queue='celery_demo'))
    print("等待10秒后查询结果")
    time.sleep(10)
    for index, task in enumerate(tasks):
        task_result = get_result(task.id)
        if task_result is not None:
            print(f"任务{index}的返回值是：{task_result}")
        else:
            print(f"任务{index}还没执行结束")
    print("再等待10秒")
    time.sleep(10)
    print(f"任务10的返回值是：{get_result(tasks[-1].id)}")


def priority_call():
    tasks = []
    print("先下发10个任务，占满worker的并发")
    for _ in range(10):
        # apply_async提供priority参数指定优先级
        tasks.append(demo_task.apply_async((_, _), routing_key='celery_demo', queue='celery_demo', priority=0))
    # 保险起见，sleep 1
    time.sleep(1)
    print("再以优先级由低到高的顺序下发10个任务，预期任务将逆序执行")
    for _ in range(10, 20):
        # apply_async提供priority参数指定优先级
        tasks.append(demo_task.apply_async((_, _), routing_key='celery_demo', queue='celery_demo', priority=_ % 10))


def chain_call():
    # 1 * 2 * 3 * 4 = 24
    # .s()是.signature(）的缩写
    # 还可通过管道符调用chain，具体参考文档
    res = chain(
        *[demo_task2.signature(_, routing_key='celery_demo', queue='celery_demo') for _ in [(1, 2), (3,), (4,)]])()
    print(res.id)
    print(f"chain任务：1 * 2 * 3 * 4={res.get()}")


def group_call():
    res = group(
        *[demo_task2.signature(_, routing_key='celery_demo', queue='celery_demo') for _ in [(1, 2), (3, 4), (5, 6)]])()
    print(res.id)
    print(f"chain任务：1 * 2, 3 * 4, 5 * 6={res.get()}")


def chord_call():
    res = chord(
        (demo_task2.signature(_, routing_key='celery_demo', queue='celery_demo') for _ in [(1, 2), (3, 4), (5, 6)]),
        tsum.s().set(routing_key='celery_demo', queue='celery_demo')
    )()
    print(res.id)
    print(f"chord任务：sum(1 * 2, 3 * 4, 5 * 6)={res.get()}")


def chunk_call():
    res = chunks(demo_task2.s(), [(1, 2), (3, 4), (5, 6)], 2).apply_async(routing_key='celery_demo', queue='celery_demo')
    print(res.id)
    print(f"chunk任务：1 * 2, 3 * 4={res.get()[0]}, 5 * 6={res.get()[1]}")


if __name__ == '__main__':
    call()
    priority_call()
    chain_call()
    group_call()
    chord_call()
    chunk_call()