更新

Author: wangchao

docs/chapter7/dispatching_a_simple_task.md

@@ -0,0 +1,175 @@
+# 分发简单任务
+
+在之前，我们已经建立好环境。下面测试一下环境，发送一个计算平方根的任务。
+
+定义任务模块`tasks.py`。在开始，导入必须的模块。
+
+```python
+from math import sqrt
+from celery import Celery
+```
+
+然后，创建`Celery`实例，代表客户端应用：
+
+```python
+app = Celery('tasks', broker='redis://192.168.25.21:6379/0')
+```
+
+在初始化时我们传入了模块的名称和`broker`的地址。
+
+然后，启动`result backend`，如下：
+
+```python
+app.config.CELERY_RESULT_BACKEND = 'redis://192.168.25.21:6379/0'
+
+# 较新的版本(v5.2.7)直接填充在celery app的初始化参数中.
+app = Celery('tasks', broker='redis://localhost/0', backend='redis://localhost/0')
+```
+
+用`@app.tack`装饰器定义任务：
+
+```python
+@app.task
+def sqrt_task(value):
+    return sqrt(value)
+```
+
+到此，我们完成了`tasks.py`模块的定义，我们需要初始化服务端的`workers`。我们创建了一个单独的目录叫做`8397_07_broker`。拷贝`tasks.py`模块到这个目录，运行如下命令：
+
+```shell
+$celery –A tasks worker –-loglevel=INFO
+```
+
+上述命令初始化了**Clery Server**，`—A`代表`Celery`应用。下图是初始化的部分截图
+
+```shell
+$# celery -A tasks worker  --loglevel=INFO
+/opt/celery_env/lib/python3.9/site-packages/celery/platforms.py:840: SecurityWarning: You're running the worker with superuser privileges: this is
+absolutely not recommended!
+
+Please specify a different user using the --uid option.
+
+User information: uid=0 euid=0 gid=0 egid=0
+
+  warnings.warn(SecurityWarning(ROOT_DISCOURAGED.format(
+
+ -------------- [email protected] v5.2.7 (dawn-chorus)
+--- ***** -----
+-- ******* ---- Linux-3.10.0-957.el7.x86_64-x86_64-with-glibc2.17 2023-03-06 16:12:10
+- *** --- * ---
+- ** ---------- [config]
+- ** ---------- .> app:         tasks:0x7fe5cbea9b80
+- ** ---------- .> transport:   redis://localhost:6379/0
+- ** ---------- .> results:     redis://localhost/0
+- *** --- * --- .> concurrency: 2 (prefork)
+-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
+--- ***** -----
+ -------------- [queues]
+                .> celery           exchange=celery(direct) key=celery
+
+
+[tasks]
+  . tasks.square_root
+
+[2023-03-06 16:12:10,866: INFO/MainProcess] Connected to redis://localhost:6379/0
+[2023-03-06 16:12:10,871: INFO/MainProcess] mingle: searching for neighbors
+[2023-03-06 16:12:11,897: INFO/MainProcess] mingle: all alone
+[2023-03-06 16:12:11,929: INFO/MainProcess] [email protected] ready.
+```
+
+现在，**Celery Server**等待接收任务并且发送给`workers`。
+
+下一步就是在客户端创建应用调用`tasks`。
+
+!!! info ""
+
+    上述步骤不能忽略，因为下面会用在之前创建的东西。
+
+在客户端机器，我们有**celery_env**虚拟环境，现在创建一个`task_dispatcher.py`模块很简单，如下步骤；
+
+1. 导入logging模块来显示程序执行信息，导入Celery模块：
+
+    ```python
+    import logging
+    from celery import Celery
+    ```
+
+2. 下一步是创建Celery实例，和服务端一样：
+
+    ```python
+    #logger configuration...
+    app = Celery('tasks', broker='redis://192.168.25.21:6379/0')
+    app.conf.CELERY_RESULT_BACKEND = 'redis://192.168.25.21:6397/0'
+    ```
+
+由于我们在接下的内容中要复用这个模块来实现任务的调用，下面我们创建一个方法来封装`sqrt_task(value)`的发送，我们将创建`manage_sqrt_task(value)`方法：
+
+```python
+def manage_sqrt_task(value):
+    result = app.send_task('tasks.sqrt_task', args=(value,))
+    logging.info(result.get())
+```
+
+从上述代码我们发现客户端应用不需要知道服务端的实现。通过**Celery**类中的`send_task`方法，我们传入`module.task`格式的字符串和以元组的方式传入参数就可以调用一个任务。最后，我们看一看`log`中的结果。
+在`__main__`中，我们调用了`manage_sqrt_task(value)`方法：
+
+```python
+if __name__ == '__main__':
+    manage_sqrt_task(4)
+```
+
+下面的截图是执行`task_dispatcher.py`文件的结果：
+
+```shell
+[2023-03-06 16:18:45,481: INFO/MainProcess] Task tasks.sqrt_task[3ecab729-f1cb-4f29-bb47-b713b2e563ed] received
+[2023-03-06 16:18:45,500: INFO/ForkPoolWorker-2] Task tasks.sqrt_task[3ecab729-f1cb-4f29-bb47-b713b2e563ed] succeeded in 0.015412827953696251s: 2.0
+```
+
+在客户端，通过`get()`方法得到结果，这是通过`send_task()`返回的`AsyncResult`实例中的重要特征。结果如下图：
+
+```shell
+$# python task_dispatcher.py
+2023-03-06 16:26:05,841 - 2.0
+```
+
+## 完整案例
+
+`tasks.py`
+
+```python
+from math import sqrt
+from celery import Celery
+
+app = Celery('tasks', broker='redis://localhost/0', backend='redis://localhost/0')
+
+
+@app.task
+def sqrt_task(value):
+    return sqrt(value)
+```
+
+`task_dispatcher.py`
+
+```python
+import logging
+from celery import Celery
+
+logger = logging.getLogger()
+logger.setLevel(logging.DEBUG)
+formatter = logging.Formatter('%(asctime)s - %(message)s')
+
+ch = logging.StreamHandler()
+ch.setLevel(logging.DEBUG)
+ch.setFormatter(formatter)
+logger.addHandler(ch)
+
+app = Celery('tasks', broker='redis://localhost/0', backend='redis://localhost/0')
+
+def manage_sqrt_task(value):
+    result = app.send_task('tasks.sqrt_task', args=(value,))
+    logger.info(result.get())
+
+
+if __name__ == '__main__':
+    print(manage_sqrt_task(4))
+```

docs/chapter7/index.md

@@ -1,12 +1,12 @@
 # 使用Celery分发任务
 
-在上一章中，我们了解并使用了 并行 Python。 我们看到了案例研究的实施，包括斐波那契数列项和使用并行 Python 模块的 Web 爬虫。 我们学习了如何使用管道在进程之间建立通信，以及如何在网络中的不同机器之间分配进程。 在本章中，我们将研究如何使用 `Celery` 框架在网络中的不同机器之间分配任务。
+在上一章中，我们了解并使用了**并行 Python**（parallel Python）。 我们看到了案例研究的实施，包括斐波那契数列项和使用并行 Python 模块的 Web 爬虫。 我们学习了如何使用管道在进程之间建立通信，以及如何在网络中的不同机器之间分配进程。 在本章中，我们将研究如何使用 Celery 框架在网络中的不同机器之间分发任务。
 
 在本章中，我们将讨论以下主题：
 
 - 理解 Celery
 - 理解 Celery 的架构
-- 设置环境
+- 搭建环境
 - 分派一个简单的任务
 - 使用 Celery 获取斐波那契数列项
 - 使用 Celery 制作分布式网络爬虫

docs/chapter7/setting_up_the_environment.md

@@ -0,0 +1,108 @@
+# 建立环境
+
+在本节中，我们将在 `Linux` 中设置两台机器。 第一个，主机名 `foshan`，将执行客户端角色，应用程序 `Celery` 将在其中调度要执行的任务。 另一台主机名为 `Phoenix` 的机器将执行**代理**（broker）、**结果后端**（result backend）和worker使用的队列的角色。
+
+## 配置客户端机器
+
+让我们开始设置客户端机器。 在这台机器上，我们将使用 `pyvenv` 工具设置一个 `Python 3.3` 的虚拟环境。 `pyvenv` 的目标是不使用额外的模块污染操作系统中存在的 `Python`，而是将每个项目所需的开发环境分开。 我们将执行以下命令来创建我们的虚拟环境：
+
+ ```shell
+ $pyvenv celery_env
+ ```
+
+上述命令在当前路径创建一个名为`celery_env`的文件夹，里面包含所有Python开发环境必须的结构。下图是该目录所包含的内容：
+
+```shell
+# 这里使用的最新的python venv模块
+$# ./Python-3.9.14/python -m venv celery_env
+$# ls celery_env/
+bin  include  lib  lib64  pyvenv.cfg
+```
+
+在创建了虚拟环境之后，我们就可以开始工作并安装需要使用的包。然而，首先我们得激活这个环境，执行以下命名：
+
+```shell
+$# source celery_env/bin/activate
+```
+
+当命令行提示符改变了，例如在左边出现`celery_env`，就说明激活完成。所有你安装的包都只在这个目录下有效，而不是在整个系统中有效。
+
+```shell
+(celery_env) $# ls celery_env/
+bin  include  lib  lib64  pyvenv.cfg
+```
+
+!!! info ""
+
+    用`--system-site-packages`标识可以创建能够访问系统`site-packages`的虚拟环境，但是不推荐使用。
+
+现在，我们有一个虚拟环境，假设已经安装好了`setuptools`或者`pip`。下面为客户端安装必须的包，如下命令：
+
+```shell
+$pip install celery
+```
+
+下图是已经安装好的framework v3.1.9，将在本书中使用该版本。
+
+```shell
+# 由于当前(2023)python2已不再支持，顾这里安装的最新版本v5.2.7
+(celery_env) $# python
+Python 3.9.14 (main, Sep 19 2022, 12:04:09)
+[GCC 4.8.5 20150623 (Red Hat 4.8.5-44)] on linux
+Type "help", "copyright", "credits" or "license" for more information.
+>>> import celery
+>>> celery.VERSION
+version_info_t(major=5, minor=2, micro=7, releaselevel='', serial='')
+>>>
+```
+
+现在我们要在**Celery**中安装支持的**Redis**，这样客户端就可以通过`broker`传输消息了。用如下命令：
+
+```shell
+$pip install celery[redis]
+```
+
+现在我们的客户端环境配置好了，在开始编码之前，我们必须配置好服务器端的环境。
+
+## 配置服务器
+
+为了配置服务器，我们首先安装**Redis**，**Redis**将作为`broker`和`result backend`。使用如下命令：
+
+```shell
+$sudo apt-get install redis-server
+```
+
+启动Redis：
+
+```shell
+$redis-server
+```
+
+如果成功，会出现类似下图中的输出
+
+```log
+2905:C 06 Mar 15:53:46.571 * supervised by systemd, will signal readiness
+                _._
+           _.-``__ ''-._
+      _.-``    `.  `_.  ''-._           Redis 3.2.12 (00000000/0) 64 bit
+  .-`` .-```.  ```\/    _.,_ ''-._
+ (    '      ,       .-`  | `,    )     Running in standalone mode
+ |`-._`-...-` __...-.``-._|'` _.-'|     Port: 6379
+ |    `-._   `._    /     _.-'    |     PID: 2905
+  `-._    `-._  `-./  _.-'    _.-'
+ |`-._`-._    `-.__.-'    _.-'_.-'|
+ |    `-._`-._        _.-'_.-'    |           http://redis.io
+  `-._    `-._`-.__.-'_.-'    _.-'
+ |`-._`-._    `-.__.-'    _.-'_.-'|
+ |    `-._`-._        _.-'_.-'    |
+  `-._    `-._`-.__.-'_.-'    _.-'
+      `-._    `-.__.-'    _.-'
+          `-._        _.-'
+              `-.__.-'
+
+2905:M 06 Mar 15:53:46.574 # WARNING: The TCP backlog setting of 511 cannot be enforced because /proc/sys/net/core/somaxconn is set to the lower value of 128.
+2905:M 06 Mar 15:53:46.574 # Server started, Redis version 3.2.12
+2905:M 06 Mar 15:53:46.574 # WARNING overcommit_memory is set to 0! Background save may fail under low memory condition. To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the command 'sysctl vm.overcommit_memory=1' for this to take effect.
+2905:M 06 Mar 15:53:46.574 # WARNING you have Transparent Huge Pages (THP) support enabled in your kernel. This will create latency and memory usage issues with Redis. To fix this issue run the command 'echo never > /sys/kernel/mm/transparent_hugepage/enabled' as root, and add it to your /etc/rc.local in order to retain the setting after a reboot. Redis must be restarted after THP is disabled.
+2905:M 06 Mar 15:53:46.574 * The server is now ready to accept connections on port 6379
+```

docs/chapter7/understanding_celery.md

@@ -0,0 +1,18 @@
+# 理解Celery
+
+**Celery** 是一个框架，该框架提供机制来简化构建分布式系统的过程。 Celery 框架通过在作为网络互连的机器或本地网络之间交换消息来使用**工作单元**（tasks）分布的概念。 任务是 **Celery** 中的关键概念； 我们必须分发的任何类型的工作都必须事先封装在任务中。
+
+## 为什么使用Celery
+
+它以透明的方式在分布在 Internet 上的工作人员或本地工作人员之间分配任务
+
+Celery有如下优点：
+
+- 它以透明的方式在网络上分布的worker或本地网络之间分配任务
+- 它通过设置（进程、线程、Gevent、Eventlet）以一种简单的方式改变了worker的并发性
+- 支持同步、异步、周期、定时任务
+- 它会在出现错误时重新执行任务
+
+!!! info ""
+
+    很多开发者都认为**同步任务**(synchronous tasks)和**实时任务**(real-time tasks)是一样的，实际上它们是完全不同的。对于**实时任务**，它有一个时间窗口，任务执行必须在`Deadline`之前完成。如果经过分析，任务在时间窗口内完成不了，那么它将被终止或者暂停直到下次能够完成，而**同步任务**是当任务执行完后才返回结果。

docs/chapter7/understanding_celery_architecture.md

@@ -0,0 +1,60 @@
+# 理解Celery架构
+
+Celery架构基于**可插拔组件**(pluggable components)和根据选择的**消息传输**（代理）(message transport(broker))协议实现的消息交换机制。下图说明了这一点：
+
+![1](../imgs/7-01.png)
+
+现在，让我们详细的介绍Celery的每个组件。
+
+## 处理任务
+
+在上图中的*client*组件，有创建和分派任务到brokers的方法。
+
+分析如下示例代码来演示通过使用`@app.task`装饰器来定义一个任务，它可以被一个**Celery**应用的实例访问，下面代码展示了一个简单的`Hello World app`：
+
+```python
+ @app.task
+ def hello_world():
+  return "Hello I'm a celery task"
+```
+
+!!! info ""
+
+    任何可执行的方法或对象都可以成为任务 (Any callable can be a task.)
+
+正如我们前面提到的，有几种类型的任务：`同步`、`异步`、`定期`和`计划`。 当我们执行任务调用时，它会返回一个 `AsyncResult` 类型的实例。 `AsyncResult` 对象是一个对象，它允许检查任务状态、它的结束，并且很明显，它在存在时返回。 但是，要使用此机制，另一个组件（结果后端）必须处于活动状态。 这将在本章中进一步解释。 要分派任务，我们应该使用任务的以下一些方法：
+
+- `delay(arg, kwarg=value)` : 这是调用 `apply_async` 方法的快捷方式。
+- `apply_async((arg,), {'kwarg': value})` : 这允许为任务的执行设置一系列有趣的参数。 其中一些如下：
+   - `countdown` : 默认任务是立即执行，该参数设置经过`countdown`秒之后执行。
+   - `expires` : 代表经过多长时间终止。
+   - `retry` : 此参数决定在连接或发送任务失败的情况下，是否必须重新发送。
+   - `queue` : 该任务所处的任务队列。
+   - `serializer` : 这表示磁盘中任务序列化的数据格式，一些示例包括 json、yaml 等。
+   - `link` : 如果发送的任务成功执行，这将链接一个或多个要执行的任务。
+   - `link_error` : 这将在任务执行失败的情况下链接一个或多个要执行的任务。
+- `apply((arg,), {'kwarg': value})` : 这会以同步方式在本地进程中执行任务，从而阻塞直到结果准备就绪为止。
+
+!!! info ""
+
+    Celery 还提供了伴随任务状态的机制，这对于跟踪和映射处理的真实状态非常有用。 有关内置任务状态的更多信息，请访问<http://celery.readthedocs.org/en/latest/reference/celery.states.html>{target="_blank"}
+
+## 理解消息转发(broker)
+
+`broker`绝对是 **Celery** 中的关键组成部分。 通过它，我们可以发送和接收消息并与`worker`沟通。 **Celery** 支持大量的代理。 然而，对于其中一些，并不是所有的 `Celery` 机制都得到了实现。 就功能而言最完整的是 `RabbitMQ` 和 `Redis`。 在本书中，我们将使用 `Redis` 作为`broker`和结果后端。 `broker`的功能是在发送任务的客户端应用程序和执行任务的工作线程之间提供一种通信方式。 这是通过使用任务队列完成的。 我们可以有几台带有代理的网络机器等待接收消息以供`workers`使用。
+
+## 理解workers
+
+`Workers`负责执行接收到的任务。**Celery**提供了一系列的机制，我们可以选择最合适的方式来控制`workers`的行为。这些机制如下：
+
+- **并发模式**(Concurrency mode)：例如**进程**、**线程**、**协程**(Eventlet)和**Gevent**。
+- **远程控制**(Remote control)：使用这种机制，可以通过高优先级队列发送消息到某个特定的`worker`来改变行为，包括在**运行时**(runtime)。
+- **撤销任务**(Revoking tasks)：使用这种机制，我们可以指示一个或多个`worker`忽略一个或多个任务的执行。
+
+如果需要，可以在运行时设置甚至更改更多功能。 比如`worker`在一段时间内执行的任务数，`worker`从哪个`queue`中消耗的时间最多等等。 有关`worker`的更多信息，请访问<http://docs.celeryproject.org/en/latest/userguide/workers.html#remote-control>{target="_blank"}
+
+## 理解result backends
+
+**结果后端**(result backend)组件的作用是存储返回给客户端应用程序的任务的状态和结果。 从 `Celery` 支持的结果后端，比较出彩的有 `RabbitMQ`、`Redis`、`MongoDB`、`Memcached` 等。 前面列出的每个**结果后端**(result backend)都有优点和缺点。 有关详细信息，请参阅 <http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-result-backends>{target="_blank"}。
+
+现在，我们对 `Celery` 架构及其组件有了一个大致的了解。 因此，让我们建立一个开发环境来实现一些例子。