# Flask實作_消息隊列_Celery_03_入門_WorkFlow ###### tags: `flask` `celery` [Celery_Next Steps](http://docs.celeryproject.org/en/latest/getting-started/next-steps.html#next-steps) 因為官方文件說明中的Next Steps擁有太多的資訊了，所以在說明記錄上拆成兩份，接續著理解Celery中的WorkFlow概念。 ## 練習 專案架構與[Flask實作_消息隊列_Celery_02_入門](/AUPlEQVnRkGSt8_ER48dZg)相同，不另外說明，可直接參閱。 ### Canvas: Designing Work-flows(工作流程) 多數情況下，我們在Calling Tasks中提到的delay與apply_async已經足夠我們應用了，但如果我們想把任務調用給另外的任務或是函數的時候，就需要用到另一個方式`signature`，signature包裝了單個參數以及執行選項再以某種方式來傳遞給函數。 ```python= >>> tasks.add.signature((2,2), countdown=10) proj.tasks.add(2, 2) >>> tasks.add.s((2,2), countdown=10) proj.tasks.add(2, 2) ``` 第1行：對add任務建立signature，並提供參數(2,2)以及執行選項`countdown=10` 第3行：可以減寫以s代替signature Signature支援呼叫api，這代表它擁有兩個method，`delay`與`apply_async`，但有一個地方不同，Signature也許已經指定參數設置。 tasks.add中需要兩個參數，兩個Signature組合成完整的Signature執行，這類似於partial function ```python= # 先宣告一個partial function(2, ) >>> s1 = tasks.add.s(2) >>> s1.values() dict_values(['proj.tasks.add', (2,), {}, {}, None, False, None]) # 再用delay加另一個參數(2, 8) >>> res = s1.delay(8) >>> res.get() 10 ``` 第2行：利用Signature加入一個參數 第3行：查詢目前s1的values 第6行：利用delay再加入一個參數 * sig.apply_async(args=(), kwargs={}, \**options) * sig.delay(*args, \**kwargs) ### The Primitives(基元?) 註：翻譯用語是在國家教育研究院查詢到的 Primitives本身就是Signature，可以用來組合成複雜的任務型態。 * Groups ```python= >>> from celery import group >>> from proj.tasks import add >>> g = group(add.s(i) for i in range(10)) >>> g group([proj.tasks.add(0), add(1), add(2), add(3), add(4), add(5), add(6), add(7), add(8), add(9)]) >>> g(10).get() [10, 11, 12, 13, 14, 15, 16, 17, 18, 19] ``` 第3行： `i for i in range(10)`這是一個標準的串列生成式，會產生0-9的值 `add.s(i) for i in range(10)`產生0-9的partial function Signature 利用`group`包裝這個Signature之後，賦值`g`，所以可以看到g本身包含的資訊 第6行：給另外一個參數的值並利用`get`取得任務結果 ```python= >>> group(add.s(i, i) for i in range(10))().get() [0, 2, 4, 6, 8, 10, 12, 14, 16, 18] ``` 第1行：將值兩兩相加 * Chains 將tasks鏈結在一起，一個完成之後呼叫另外一個 ```python= >>> from celery import chain >>> from proj.tasks import add, mul >>> chain(add.s(4,4) | mul.s(8))().get() 64 ``` 第3行：將4+4=8的結果傳到下一個任務中與8相乘，即8x8=64 ```python= >>> g = chain(add.s(4) | mul.s(8)) >>> g(4).get() 64 ``` 第1行：利用partial function實作 ```python= >>> (add.s(4, 4) | mul.s(8))().get() ``` 第1行：也可以直接鏈結 * Chords Chords就是Group帶有Callback ```python= >>> from celery import chord >>> from proj.tasks import add, xsum >>> chord((add.s(i, i) for i in range(10)), xsum.s())().get() ``` 當Group通過Chain，就會變成是Chords ```python= >>> (group(add.s(i, i) for i in range(10)) | xsum.s())().get() ``` ### Routing 預設上，Celery在派送任務的時候所使用的Routing是celery，但這部份可以利用配置來設置，後續再 使用`apply_async`的時候再宣告要走的Queue ```python= app.conf.update( task_routes = { 'proj.tasks.add': {'queue': 'hipri'}, }, ) ``` 利用參數設置我們新增了一個`hipri`的隊列，這可以讓我們在後續指定調用。 ```python= >>> from proj.tasks import add >>> add.apply_async((2, 2), queue='hipri') ``` 也可以在啟動的時候指定要處理的隊列名稱 ```python= celery -A proj worker -Q hipri,celery ``` [Routing](http://docs.celeryproject.org/en/latest/userguide/routing.html#guide-routing) ### 時區設置 ```python= app.conf.timezone = 'Europe/London' ``` 記得調整成自己國家的就可以了 ## 總結 按著官方教學說明我們認識了Signature以及六種Primitives中的三種，其餘三種可以直接再參考官方說明，也了解到怎麼設置Routing(對應RabbitMQ的Queue)。 到這邊，也算是有基礎了，剩下的就是實作中學習了。