Add chapter 7

Author: liufei01

第七章/分发简单任务.md

@@ -0,0 +1,70 @@
+在之前，我们已经建立好环境。下面测试一下环境，发送一个计算平方根的任务。
+定义任务模块tasks.py。在开始，导入必须的模块。
+
+	from math import sqrt
+	from celery import Celery
+
+然后，创建Celery实例，代表客户端应用：
+
+	app = Celery('tasks', broker='redis://192.168.25.21:6379/0')
+
+在初始化时我们传入了模块的名称和broker的地址。
+然后，启动result backend，如下：
+
+	app.config.CELERY_RESULT_BACKEND = 'redis://192.168.25.21:6379/0'
+
+用@app.tack装饰器定义任务：
+
+	@app.task
+	def square_root(value):
+		return sqrt(value)
+
+到此，我们完成了tasks.py模块的定义，我们需要初始化服务端的workers。我们创建了一个单独的目录叫做8397_07_broker。拷贝tasks.py模块到这个目录，运行如下命令：
+
+	$celery –A tasks worker –-loglevel=INFO
+
+上述命令初始化了Clery Server，—A代表Celery应用。下图是初始化的部分截图
+
+![](图片链接地址)
+
+现在，Celery Server等待接收任务并且发送给workers。
+下一步就是在客户端创建应用调用tasks。
+ 
+ > 上述步骤不能忽略，因为下面会用在之前创建的东西。
+
+在客户端机器，我们有celery_env虚拟环境，现在创建一个task_dispatcher.py模块很简单，如下步骤；
+1. 导入logging模块来显示程序执行信息，导入Celery模块：
+
+	import logging
+	from celery import Celery
+
+2. 下一步是创建Celery实例，和服务端一样：
+
+	#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)方法：
+
+	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)方法：
+
+	if __name__ == '__main__':
+		manage_sqrt_task(4)
+
+下面的截图是执行task_dispatcher.py文件的结果：
+
+![](图片链接地址)
+
+在客户端，通过get()方法得到结果，这是通过send_task()返回的AsyncResult实例中的重要特征。结果如下图：
+
+![](图片链接地址)
+
+

第七章/建立环境.md

@@ -0,0 +1,51 @@
+### 建立环境
+
+在本章，我们将使用两台linux机器。第一台主机名foshan，作为客户端，执行app Celery分发的任务。另一台机器主机名Phoenix，作为broker、result backend和任务队列。
+
+### 配置客户端机器
+
+首先配置客户端机器。在这台机器上，用pyvenv建立Python3.3的虚拟环境。用pyvenv的目的是隔离每个项目的开发环境。执行以下命令能够创建虚拟环境：
+
+	$pyvenv celery_env
+
+上述命令在当前路径创建一个名为celery_env的文件夹，里面包含所有Python开发环境必须的结构。下图是该目录所包含的内容：
+
+![](图片链接地址)
+
+在创建了虚拟环境之后，我们就可以开始工作并安装需要使用的包。然而，首先我们得激活这个环境，执行以下命名：
+	
+	$source bin/activate
+
+当命令行提示符改变了，例如在左边出现celery_env，就说明激活完成。所有你安装的包都只在这个目录下有效，而不是在整个系统中有效。
+
+> 用--system-site-packages标识可以创建能够访问系统site-packages的虚拟环境，但是不推荐使用。
+
+现在，我们有一个虚拟环境，假设已经安装好了setuptools或者pip。下面为客户端安装必须的包，如下命令：
+
+	$pip install celery
+
+下图是已经安装好的framework v3.1.9，将在本书中使用该版本。
+
+![](图片链接地址)
+
+现在我们要在Celery中安装支持的Redis，这样客户端就可以通过broker传输消息了。用如下命令：
+
+	$pip install celery[redis]
+
+现在我们的客户端环境配置好了，在开始编码之前，我们必须配置好服务器端的环境。
+
+
+### 配置服务器
+
+为了配置服务器，我们首先安装Redis，Redis将作为broker和result backend。使用如下命令：
+
+	$sudo apt-get install redis-server
+
+启动Redis：
+	
+	$redis-server
+
+如果成功，会出现类似下图中的输出
+
+![](图片链接地址)
+

第七章/总结.md

@@ -0,0 +1,2 @@
+在本章，我们论述了Celery的分布式任务队列，并且对Celery的架构、组件、基本环境搭建、运行简单应用有了了解。如果单独讲Celery可以写一本书，通过本章需要你对Celery有基本的认识。
+在下一章，我们将学习asyncio模块和怎样以异步的方式执行程序。下一章也会涉及到协程(coroutines)和asyncio中协程的应用。

第七章/根据任务类型定义队列.md

@@ -0,0 +1,28 @@
+计算斐波那契数列的任务已经实现并且运行。所有任务都被发送到默认的Celery队列。然而，有方法路由任务到不同的队列。让我们重构服务器端的架构来实现路由任务。我们将根据任务类型定义队列。
+在服务器端启动Celery server时，我们将建立三种不同的队列，它们将被workers消费。斐波那契数列任务有fibo_queue，平方根任务有sqrt_queue，网络爬虫任务有 webcrawler_queue。然而，划分队列有什么好处呢。以下列举一些：
+
+- 相同类型的任务划分为一组，使得监控更加简单
+- 定义worker从一个队列取任务，性能更佳
+- 可在性能更好的机器上建立任务更重的队列
+
+为了再server上建立队列，我们只需用如下命令初始化Celery：
+
+	$celery –A tasks –Q sqrt_queue,fibo_queue,webcrawler_queue worker --loglevel=info
+
+下图是在服务端截图：
+
+![](图片链接地址)
+
+在进行下一个例子之前，我们路由现有的任务到队列中。在服务端，在task_dispatcher.py模块中，我们将修改send_task调用来将任务发送到不同的队列。修改如下：
+
+	app.send_task('tasks.sqrt_task', args=(value,),
+		queue='sqrt_queue', routing_key='sqrt_queue')
+
+然后，修改fibo_task调用，如下：
+
+	app.send_task('tasks.fibo_task', args=(x,), queue='fibo_queue',
+		routing_key='fibo_queue')
+
+> 如果有兴趣监控队列、统计任务数量或者其他，请参考Celery文档 http://celery.readthedocs.org/en/latest/userguide/monitoring.html。
+> 在任何情况用Redis，redis-cli都可以作为一个工具。队列、任务、workders都可以被监控，详见 http://celery.readthedocs.org/en/latest/userguide/monitoring.html#workers.
+

第七章/理解Celery.md

@@ -0,0 +1,11 @@
+Celery是一个框架，该框架提供机制来简化构建分布式系统的过程。Celery框架通过在同一个网络内的机器之间交换信息来分发任务。任务(task)在Celery中是一个关键的概念，任何形式的job在分发之前都必须封装成任务。
+
+## 为什么使用Celery
+
+Celery有如下优点：
+- 分发任务是透明的
+- 在并发worker启动时，对其改变很小
+- 支持同步、异步、周期和计划任务
+- 如果出现了错误，会重新执行任务
+
+> 很多开发者都认为同步任务和实时任务是一样的，实际上它们是完全不同的。对于实时任务，它有一个时间窗口，任务执行必须在Deadline之前完成。如果经过分析，任务在时间窗口内完成不了，那么它将被终止或者暂停直到下次能够完成，而同步任务是当任务执行完后才返回结果。

第七章/理解Celery架构.md

@@ -0,0 +1,50 @@
+Celery架构基于可插拔组件和用*message transport(broker)*实现的消息交换机制。具体的如下图所示：
+![](图片链接地址)
+
+现在，让我们详细的介绍Celery的每个组件。
+
+### 处理任务
+
+在上图中的*client*组件，有创建和分派任务到brokers的方法。
+分析如下示例代码来演示通过使用@app.task装饰器来定义一个任务，它可以被一个Celery应用的实例访问，下面代码展示了一个简单的`Hello World app`：
+
+	@app.task
+	def hello_world():
+		return "Hello I'm a celery task"
+
+
+> 任何可以被调用的方法或对象都可以成为任务
+
+正如我们之前所说，有各种类型的任务：同步、异步、周期和计划任务。当我们调用任务，它返回类型AsyncResult的实例。AsyncResult对象中可以查看任务状态，当任务结束后可以查看返回结果。然而，为了利用这个机制，另一个叫做*result backend*的组件必须启动，这将在本章的后面讲解。为了分派任务，我们可以用下面这些方法：
+- delay(arg, kwarg=value) : 它会调用apply_aync方法。
+- apply_async((arg,), {'kwarg': value}) : 该方法可以为任务设置很多参数，一些参数如下。
+ - countdown : 默认任务是立即执行，该参数设置经过countdown秒之后执行。
+ - expires : 代表经过多长时间终止。
+ - retry : 如果连接或者发送任务失败，该参数可是重试。
+ - queue : 任务队列。
+ - serializer : 数据格式，其他还有json、yaml等等
+ - link : 连接一个或多个即将执行的任务。
+ - link_error : 连接一个或多个执行失败的任务。
+- apply((arg,), {'kwarg': value}) : 在本地进程以同步的方式执行任务，因而阻塞直到结果就绪。
+
+> Celery 提供了查看任务状态的机制，这在跟踪进程的真实状态非常有用。更多的关于内建任务状态的资料请查看http://celery.readthedocs.org/en/latest/reference/celery.states.html
+
+### 发现消息传输(broker)
+
+broker是Celery中的核心组件，通过broker可以发送和接受消息，来完成和workers的通信。Celery支持大量的brokers。然而，对于某些broker，不是所有的Celery机制都实现了。实现功能最全的是**RabbitMQ**和**Redis**。在本书中，我们将采用Redis作为broker。broker提供在不同客户端应用之间通信的方法，客户端应用发送任务，workers执行任务。可以有多台带有broker的机器等待接收消息，然后发送消息给workers。
+
+
+### 理解workers
+
+Workers负责执行接收到的任务。Celery提供了一系列的机制，我们可以选择最合适的方式来控制workers的行为。这些机制如下：
+- 并发模式：例如进程、线程、协程(Eventlet)和Gevent
+- 远程控制：利用该机制，可以通过高优先级队列发送消息到某个特定的worker来改变行为，包括runtime。
+- 撤销任务：利用该机制，可以指挥一个或多个workers来忽略一个或多个任务。
+
+更多的特性可以在运行时设定或者改变。比如，worker在某一段时间执行的任务数，worker在哪个队列消费等等。软玉worker更多的信息可以参考  http://docs.celeryproject.org/en/latest/userguide/workers.html#remote-control 
+
+### 理解result backends
+
+Result backend组件存储任务的状态和任务返回给客户端应用的结果。Celery支持的result backend之中，比较出彩的有 RabbitMQ, Redis, MongoDB, Memcached。每个result backend都有各自的优缺点，详见  http://docs.celeryproject.org/en/latest/userguide/tasks.html#task-result-backends
+
+现在，我们对Celery架构有了一个大概的认识。下面我们建立一个开发环境来实现一些例子。

第七章/用Celery来分发任务.md

@@ -0,0 +1,8 @@
+在之前的章节，我们学习了Python并行编程。我们用Python并行模块实现了一些实例，包括斐波那契数列和网络爬虫。我们知道了怎样用管道进行进程间通信和如何在同一个网络的不同机器上分发进程。在本章节，我们将学习如何利用Celery框架在同一个网络的不同机器之间分发任务。
+在本章，我们将覆盖一下几个主题：
+- 理解Celery
+- 理解Celery架构
+- 建立环境
+- 分发简单任务
+- 用Celery来获得斐波那契数列的项
+- 用Celery来构建一个分布式网络爬虫系统

第七章/用Celery来构建一个分布式网络爬虫系统.md

@@ -0,0 +1,65 @@
+现在我们将用Celery构建网络爬虫。我们已经有了webcrawler_queue，负责hcrawler任务。然而，在服务器端，我们将在tasks.py模块创建crawl_task任务。
+首先，导入re(正则表达式)和requests(HTTP lib)模块，代码如下：
+
+	import re
+	import requests
+
+然后，定义正则表达式，和之前的章节一样；
+
+	hTML_link_regex = re.compile(
+	'<a\s(?:.*?\s)*?href=[\'"](.*?)[\'"].*?>')
+
+然后，替换crawl_task方法，添加@app.task装饰器，修改返回信息，如下：
+
+	@app.task
+	def crawl_task(url):
+		request_data = requests.get(url)
+		links = html_link_regex.findall(request_data.text)
+		message = "The task %s found the following links %s.."\
+		Return message
+
+links列表不一定要和下图匹配：
+
+![](图片链接地址)
+
+在客户端task_dipatcher.py模块实现crawl_task调用。
+首先，我们需要列出数据的输入url_list。代码如下：
+
+
+	url_list = ['http://www.google.com',
+				'http://br.bing.com',
+				'http://duckduckgo.com',
+				'http://github.com',
+				'http://br.search.yahoo.com']
+
+创建manage_crawl_task方法。
+
+	def manage_crawl_task(url_list):
+		async_result_dict = {url: app.send_task('tasks.crawl_task',
+			args=(url,), queue='webcrawler_queue',
+			routing_key='webcrawler_queue') for url in url_list}
+		for key, value in async_result_dict.items():
+			if value.ready():
+				logger.info("%s -> %s" % (key, value.get()))
+			else:
+				logger.info("The task [%s] is not ready" %
+	value.task_id)
+
+和之前创建的manage_fibo_task方法一样，async_result_dict字典包含当前URL和AsyncResult结果。然后我们检查任务的状态获取任务结果。
+
+现在我们在__main__中调用该方法：
+
+	if __main__ == '__main__':
+		#manage_sqrt_task(4)
+		#manage_fibo_task(input_list)
+		manage_crawl_task(url_list)
+
+运行task_dispatcher.py代码，在服务器端有如下输出：
+
+![](图片链接地址)
+
+最后，客户端的输出如下：
+
+![](图片链接地址)
+
+Celery是一个强大的工具，在本章我们只是用到了基本的东西。更多的内容建议自己在真实的项目中动手去尝试。

第七章/用Celery来获得斐波那契数列的项.md

@@ -0,0 +1,58 @@
+让我们再一次计算多个输入的斐波那契数列的项，每个都用分布式的方法。现在的方法和之前的方法改变很小。
+我们将终止Celery的执行(Ctrl+C)并且在tasks.py模块(之前创建过)加入fibo_task任务。
+
+	@app.task
+	def fibo_task(value):
+		a, b = 0,1
+		for item in range(value):
+			a, b = b, a + b
+		message = "The Fibonacci calculated with task id %s" \
+		" was %d" % (fibo_task.request.id, a)
+		Return (value, message)
+
+通过\<task.reaquest.id\>得到任务的ID，请求对象是task的对象，task对象提供了task执行的上下文。通过上下文可以得到task的ID等信息。
+
+在tasks.py模块加入了新的任务之后，再一次初始化Celery，结果如下图：
+
+![](图片链接地址)
+
+现在我们把fibo_task任务装载到Celery server，我们将在客户端实现对该任务的调用。
+
+在task_dispatcher.py模块，我们会申明input_list，如下：
+
+	input_list = [4, 3, 8, 6, 10]
+
+和前面的做法一样，定义manage_fibo_task方法：
+
+	def manage_fibo_task(value_list):
+		async_result_dict = {x: app.send_task('tasks.fibo_task',
+			args=(x,)) for x in value_list}
+		for key, value in async_result_dict.items():
+			logger.info("Value [%d] -> %s" % (key, value.get()[1]))
+
+在manage_fibo_task方法中，创建了一个叫做async_result_dict的字典，key是传入的要计算的值，value是send_task方法返回的AyncResult对象。通过这个方法，我们可以查看任务的结果和状态。
+最后，遍历字典得到输入值和输出结果。AsyncResult的get方法能够得到结果。
+get()方法会阻塞进程。一个好的方法是调用ready()方法来检查结果是否返回了。
+上述循环可以修改为如下；
+
+	for key, value in async_result_dict.items():
+		if value.ready():
+			logger.info("Value [%d] -> %s" % (key, value.get()[1]))
+		else:
+			logger.info("Task [%s] is not ready" % value.task_id)
+
+不同的任务会有不同的延迟时间，为了防止无限等待，可以用get(timeout=x)方法设置超时。
+最后，添加manage_fibo_task的调用，传入input_list。代码如下：
+
+	if __name__ == '__main__':
+		#manage_sqrt_task(4)
+		manage_fibo_task(input_list)
+
+当我们执行task_dispatcher.py后，输入如下：
+
+![](图片链接地址)
+
+在客户端有如下输出：
+
+![](图片链接地址)
+