Python开发【模块】：Celery 分布式异步消息任务队列

「爱情、让人受尽委屈。」 2022-03-27 10:08 263阅读 0赞

**前言：**

Celery 是一个 基于python开发的**分布式异步消息任务队列**，通过它可以轻松的实现任务的异步处理， 如果你的业务场景中需要用到**异步任务**，就可以考虑使用celery， 举几个实例场景中可用的例子:

*  你想对100台机器执行一条批量命令，可能会花很长时间 ，但你不想让你的程序等着结果返回，而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果， 在任务执行ing进行时，你可以继续做其它的事情。
 *  你想做一个定时任务，比如每天检测一下你们所有客户的资料，如果发现今天 是客户的生日，就给他发个短信祝福

**Celery有以下优点：**

*  简单：一单熟悉了celery的工作流程后，配置和使用还是比较简单的
 *  高可用：当任务执行失败或执行过程中发生连接中断，celery 会自动尝试重新执行任务
 *  快速：一个单进程的celery每分钟可处理上百万个任务
 *  灵活： 几乎celery的各个组件都可以被扩展及自定制

Celery基本工作流程图：

![997909-20170710111641368-1420822519.png][]

**1、 Celery安装使用**

Celery需要在linux的环境下运行：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div> 
    <div>
      5 
    </div> 
    <div>
      6 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code># 安装</code> 
     </div> 
     <div> 
      <code>[root@localhost celerys]</code> 
      <code># pip3 install celery</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code># 进入python import无异常表示安装成功</code> 
     </div> 
     <div> 
      <code>[root@localhost celerys]</code> 
      <code># python3</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt;&nbsp;</code> 
      <code>import</code>&nbsp; 
      <code>celery</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

Celery的默认broker是RabbitMQ, 仅需配置一行就可以

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code>broker_url&nbsp;</code> 
      <code>=</code>&nbsp; 
      <code>'amqp://guest:guest@localhost:5672//'</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

使用Redis做broker也可以

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code>broker_url&nbsp;</code> 
      <code>=</code>&nbsp; 
      <code>'redis://localhost:6379/0'</code> 
     </div> 
     <div> 
      <code>#redis://:password@hostname:port/db_number</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

**2、简单使用**

创建一个任务文件就叫tasks.py:

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div> 
    <div>
      5 
    </div> 
    <div>
      6 
    </div> 
    <div>
      7 
    </div> 
    <div>
      8 
    </div> 
    <div>
      9 
    </div> 
    <div>
      10 
    </div> 
    <div>
      11 
    </div> 
    <div>
      12 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code>from</code>&nbsp; 
      <code>celery&nbsp;</code> 
      <code>import</code>&nbsp; 
      <code>Celery</code> 
     </div> 
     <div> 
      <code>import</code>&nbsp; 
      <code>time</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code>app&nbsp;</code> 
      <code>=</code>&nbsp; 
      <code>Celery(</code> 
      <code>'cly'</code> 
      <code>,&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 任意</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>broker</code> 
      <code>=</code> 
      <code>'redis://192.168.1.166:6379/0'</code> 
      <code>,&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 中间件</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>backend</code> 
      <code>=</code> 
      <code>'redis://localhost'</code> 
      <code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 数据存储</code> 
     </div> 
     <div> 
      <code>&nbsp;</code>&nbsp; 
     </div> 
     <div> 
      <code>@app</code> 
      <code>.task</code> 
     </div> 
     <div> 
      <code>def</code>&nbsp; 
      <code>add(x,y):</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>time.sleep(</code> 
      <code>10</code> 
      <code>)</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>print</code> 
      <code>(</code> 
      <code>"running..."</code> 
      <code>,x,y)</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>return</code>&nbsp; 
      <code>x</code> 
      <code>+</code> 
      <code>y</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

启动Celery Worker来开始监听并执行任务：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div> 
    <div>
      5 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code># 加入环境变量</code> 
     </div> 
     <div> 
      <code>[root@localhost ~]</code> 
      <code># PATH=$PATH:/usr/local/python3.5/bin/</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code># 启动一个worker</code> 
     </div> 
     <div> 
      <code>[root@localhost celerys]</code> 
      <code># celery -A tasks worker --loglevel=info</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

调用任务：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div> 
    <div>
      5 
    </div> 
    <div>
      6 
    </div> 
    <div>
      7 
    </div> 
    <div>
      8 
    </div> 
    <div>
      9 
    </div> 
    <div>
      10 
    </div> 
    <div>
      11 
    </div> 
    <div>
      12 
    </div> 
    <div>
      13 
    </div> 
    <div>
      14 
    </div> 
    <div>
      15 
    </div> 
    <div>
      16 
    </div> 
    <div>
      17 
    </div> 
    <div>
      18 
    </div> 
    <div>
      19 
    </div> 
    <div>
      20 
    </div> 
    <div>
      21 
    </div> 
    <div>
      22 
    </div> 
    <div>
      23 
    </div> 
    <div>
      24 
    </div> 
    <div>
      25 
    </div> 
    <div>
      26 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code>[root@localhost celerys]</code> 
      <code># python3</code> 
     </div> 
     <div> 
      <code>Python&nbsp;</code> 
      <code>3.5</code> 
      <code>.</code> 
      <code>2</code>&nbsp; 
      <code>(default, Jul&nbsp;&nbsp;</code> 
      <code>7</code>&nbsp; 
      <code>2017</code> 
      <code>,&nbsp;</code> 
      <code>23</code> 
      <code>:</code> 
      <code>36</code> 
      <code>:</code> 
      <code>01</code> 
      <code>)</code> 
     </div> 
     <div> 
      <code>[GCC&nbsp;</code> 
      <code>4.8</code> 
      <code>.</code> 
      <code>5</code>&nbsp; 
      <code>20150623</code>&nbsp; 
      <code>(Red Hat&nbsp;</code> 
      <code>4.8</code> 
      <code>.</code> 
      <code>5</code> 
      <code>-</code> 
      <code>11</code> 
      <code>)] on linux</code> 
     </div> 
     <div> 
      <code>Type</code>&nbsp; 
      <code>"help"</code> 
      <code>,&nbsp;</code> 
      <code>"copyright"</code> 
      <code>,&nbsp;</code> 
      <code>"credits"</code>&nbsp; 
      <code>or</code>&nbsp; 
      <code>"license"</code>&nbsp; 
      <code>for</code>&nbsp; 
      <code>more information.</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt;&nbsp;</code> 
      <code>from</code>&nbsp; 
      <code>tasks&nbsp;</code> 
      <code>import</code>&nbsp; 
      <code>add&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># import add</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; add.delay(</code> 
      <code>4</code> 
      <code>,</code> 
      <code>6</code> 
      <code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 执行函数</code> 
     </div> 
     <div> 
      <code>&lt;AsyncResult:&nbsp;</code> 
      <code>4b5a8ab6</code> 
      <code>-</code> 
      <code>693c</code> 
      <code>-</code> 
      <code>4ce5</code> 
      <code>-</code> 
      <code>b779</code> 
      <code>-</code> 
      <code>305cfcdf70cd</code> 
      <code>&gt;&nbsp;&nbsp;&nbsp;</code> 
      <code># 返回taskid</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result&nbsp;</code> 
      <code>=</code>&nbsp; 
      <code>add.delay(</code> 
      <code>4</code> 
      <code>,</code> 
      <code>6</code> 
      <code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 执行函数</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result.get()&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 同步获取结果,一直等待</code> 
     </div> 
     <div> 
      <code>10</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result.get(timeout</code> 
      <code>=</code> 
      <code>1</code> 
      <code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 设置超时时间，过期错误异常</code> 
     </div> 
     <div> 
      <code>Traceback (most recent call last):</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>-</code> 
      <code>-</code> 
      <code>strip</code> 
      <code>-</code> 
      <code>-</code> 
     </div> 
     <div> 
      <code>celery.exceptions.TimeoutError: The operation timed out.</code> 
     </div> 
     <div>
       &nbsp; 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result&nbsp;</code> 
      <code>=</code>&nbsp; 
      <code>add.delay(</code> 
      <code>4</code> 
      <code>,</code> 
      <code>'a'</code> 
      <code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 执行错误命令</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result.get()&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># get后获取到错误信息，触发异常</code> 
     </div> 
     <div> 
      <code>Traceback (most recent call last):</code> 
     </div> 
     <div> 
      <code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code>-</code> 
      <code>-</code> 
      <code>strip</code> 
      <code>-</code> 
      <code>-</code> 
     </div> 
     <div> 
      <code>celery.backends.base.TypeError: unsupported operand&nbsp;</code> 
      <code>type</code> 
      <code>(s)&nbsp;</code> 
      <code>for</code>&nbsp; 
      <code>+</code> 
      <code>:&nbsp;</code> 
      <code>'int'</code>&nbsp; 
      <code>and</code>&nbsp; 
      <code>'str'</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result&nbsp;</code> 
      <code>=</code>&nbsp; 
      <code>add.delay(</code> 
      <code>4</code> 
      <code>,</code> 
      <code>'a'</code> 
      <code>)</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result.get(propagate</code> 
      <code>=</code> 
      <code>False</code> 
      <code>)&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># propagate=False 不触发异常，获取错误信息</code> 
     </div> 
     <div> 
      <code>TypeError(</code> 
      <code>"unsupported operand type(s) for +: 'int' and 'str'"</code> 
      <code>,)</code> 
     </div> 
     <div> 
      <code>&gt;&gt;&gt; result.traceback&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 获取具体错误信息 log打印用</code> 
     </div> 
     <div> 
      <code>'Traceback (most recent call last):\n&nbsp; File "/usr/local/python3.5/lib/python3.5/site-packages/celery/app/trace.py", line 367, in trace_task\n&nbsp;&nbsp;&nbsp; R = retval = fun(*args, **kwargs)\n&nbsp; File "/usr/local/python3.5/lib/python3.5/site-packages/celery/app/trace.py", line 622, in __protected_call__\n&nbsp;&nbsp;&nbsp; return self.run(*args, **kwargs)\n&nbsp; File "/data/celerys/tasks.py", line 12, in add\n&nbsp;&nbsp;&nbsp; return x+y\nTypeError: unsupported operand type(s) for +: \'int\' and \'str\'\n'</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

此时worker端收到的信息：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code>[</code> 
      <code>2017</code> 
      <code>-</code> 
      <code>07</code> 
      <code>-</code> 
      <code>08</code>&nbsp; 
      <code>03</code> 
      <code>:</code> 
      <code>12</code> 
      <code>:</code> 
      <code>22</code> 
      <code>,</code> 
      <code>565</code> 
      <code>: WARNING</code> 
      <code>/</code> 
      <code>PoolWorker</code> 
      <code>-</code> 
      <code>1</code> 
      <code>] running...&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;</code> 
      <code># 获取到任务</code> 
     </div> 
     <div> 
      <code>[</code> 
      <code>2017</code> 
      <code>-</code> 
      <code>07</code> 
      <code>-</code> 
      <code>08</code>&nbsp; 
      <code>03</code> 
      <code>:</code> 
      <code>12</code> 
      <code>:</code> 
      <code>22</code> 
      <code>,</code> 
      <code>565</code> 
      <code>: WARNING</code> 
      <code>/</code> 
      <code>PoolWorker</code> 
      <code>-</code> 
      <code>1</code> 
      <code>]&nbsp;</code> 
      <code>4</code> 
     </div> 
     <div> 
      <code>[</code> 
      <code>2017</code> 
      <code>-</code> 
      <code>07</code> 
      <code>-</code> 
      <code>08</code>&nbsp; 
      <code>03</code> 
      <code>:</code> 
      <code>12</code> 
      <code>:</code> 
      <code>22</code> 
      <code>,</code> 
      <code>565</code> 
      <code>: WARNING</code> 
      <code>/</code> 
      <code>PoolWorker</code> 
      <code>-</code> 
      <code>1</code> 
      <code>]&nbsp;</code> 
      <code>6</code>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 
      <code># 任务执行完毕数据存储到backend端</code> 
     </div> 
     <div> 
      <code>[</code> 
      <code>2017</code> 
      <code>-</code> 
      <code>07</code> 
      <code>-</code> 
      <code>08</code>&nbsp; 
      <code>03</code> 
      <code>:</code> 
      <code>12</code> 
      <code>:</code> 
      <code>22</code> 
      <code>,</code> 
      <code>567</code> 
      <code>: INFO</code> 
      <code>/</code> 
      <code>PoolWorker</code> 
      <code>-</code> 
      <code>1</code> 
      <code>] Task tasks.add[</code> 
      <code>683e395e</code> 
      <code>-</code> 
      <code>48b9</code> 
      <code>-</code> 
      <code>4d32</code> 
      <code>-</code> 
      <code>b3bb</code> 
      <code>-</code> 
      <code>1492c62af393</code> 
      <code>] succeeded&nbsp;</code> 
      <code>in</code>&nbsp; 
      <code>10.01260852499945s</code> 
      <code>:&nbsp;</code> 
      <code>10</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

查看broker(即192.168.1.166)端数据：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> 
    <div>
      4 
    </div> 
    <div>
      5 
    </div> 
    <div>
      6 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code>[root@localhost redis</code> 
      <code>-</code> 
      <code>3.0</code> 
      <code>.</code> 
      <code>6</code> 
      <code>]</code> 
      <code># src/redis-cli</code> 
     </div> 
     <div> 
      <code>127.0</code> 
      <code>.</code> 
      <code>0.1</code> 
      <code>:</code> 
      <code>6379</code> 
      <code>&gt; keys&nbsp;</code> 
      <code>*</code> 
     </div> 
     <div> 
      <code>1</code> 
      <code>)&nbsp;</code> 
      <code>"_kombu.binding.celeryev"</code> 
     </div> 
     <div> 
      <code>2</code> 
      <code>)&nbsp;</code> 
      <code>"unacked_mutex"</code> 
     </div> 
     <div> 
      <code>3</code> 
      <code>)&nbsp;</code> 
      <code>"_kombu.binding.celery.pidbox"</code> 
     </div> 
     <div> 
      <code>4</code> 
      <code>)&nbsp;</code> 
      <code>"_kombu.binding.celery"</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

执行完后，backend端的数据：

<table> 
 <tbody> 
  <tr> 
   <td> 
    <div>
      1 
    </div> 
    <div>
      2 
    </div> 
    <div>
      3 
    </div> </td> 
   <td> 
    <div> 
     <div> 
      <code>[root@localhost redis</code> 
      <code>-</code> 
      <code>3.0</code> 
      <code>.</code> 
      <code>6</code> 
      <code>]</code> 
      <code># src/redis-cli&nbsp;&nbsp; # 程序get后,数据未被删除</code> 
     </div> 
     <div> 
      <code>127.0</code> 
      <code>.</code> 
      <code>0.1</code> 
      <code>:</code> 
      <code>6379</code> 
      <code>&gt; keys&nbsp;</code> 
      <code>*</code> 
     </div> 
     <div> 
      <code>1</code> 
      <code>)&nbsp;</code> 
      <code>"celery-task-meta-683e395e-48b9-4d32-b3bb-1492c62af393"</code> 
     </div> 
    </div> </td> 
  </tr> 
 </tbody> 
</table>

[997909-20170710111641368-1420822519.png]: /images/20220327/a0712c3f29834e6aa72e57993ef34609.png