Do you want to pick up from where you left of?
Take me there

GenStage

在本課程中，將仔細研究 GenStage、它的作用和如何在應用程式中利用。

Table of Contents

簡介

那什麼是 GenStage？從官方文件來看，它是＂Elixir 的規範 (specification) 和計算流程 (computational flow)＂，但這表示著什麼？

這表示著 GenStage 提供了一種方式，在一個單獨過程中通過獨立 steps (或 stages) 來定義管線 (pipeline) 的工作以便執行它；如果之前使用過管線，那麼其中一些概念應該不陌生。

為了更好地理解它的工作原理，現在將一個簡易的生產者-消費者 (producer-consumer) 流程可視化：

[A] -> [B] -> [C]

在這個範例中，有三個 stages：A 消費者 (producer)，B 生產者-消費者 (producer-consumer) 與 C 消費者 (consumer)。 A 產生一個被 B 消耗的值，B 執行一些工作並回傳將被 consumer C 收到的新值；我們將在下一節中看到，stage 的角色很重要。

雖然範例是 1 對 1 的生產者-消費者，但在任何特定 stage 都可能有多個生產者和多個消費者。

為了更好地描繪這些概念，將使用 GenStage 打造一個管線 (pipeline)，但首先探索一下 GenStage 會依賴頗多的這些角色。

消費者 (Consumers) 與生產者 (Producers)

正如所讀到的，我們賦予 stage 的角色非常重要。 GenStage 的規範中承認三種角色：

注意到生產者是 等待 需求了嗎？通過 GenStage，消費者是向上游發送需求並處理來自生產者的資料。 這有助於稱為背壓（back-pressure）的機制。 當消費者忙碌時，背壓機制使生產者不會造成超壓（over-pressure）。

現在已經介紹過 GenStage 中的角色，開始建立應用程式。

入門

在這個範例中，將建立一個產出數字的 GenStage 應用程式，能排序偶數後輸出。

在應用程式中，將使用全部三個 GenStage 角色。 生產者將負責計算和產出數字。 生產者－消費者則過濾出偶數，而後對下游需求做出回應。 最後建立消費者顯示剩餘數字。

讓我們從生成一個有 supervision 樹的專案開始：

mix new genstage_example --sup
cd genstage_example

現在更新 mix.exs 中的耦合性 (dependencies) 以包含 gen_stage：

defp deps do
  [
    {:gen_stage, "~> 1.0.0"},
  ]
end

在進一步研究之前，應該先取得 (fetch) 耦合關係 (dependencies) 並進行編譯：

mix do deps.get, compile

現在準備好建立生產者了！

生產者

GenStage 應用程式的第一件事是建立生產者。 正如之前討論過的，我們想要建立一個能夠產出持續不斷數字流的生產者。 現在來建立生產者資料夾和檔案：

mkdir lib/genstage_example
touch lib/genstage_example/producer.ex

這時可以加入我們所需程式碼：

defmodule GenstageExample.Producer do
  use GenStage

  def start_link(initial \\ 0) do
    GenStage.start_link(__MODULE__, initial, name: __MODULE__)
  end

  def init(counter), do: {:producer, counter}

  def handle_demand(demand, state) do
    events = Enum.to_list(state..(state + demand - 1))
    {:noreply, events, state + demand}
  end
end

這裡需要注意兩個最重要部分是 init/1 和 handle_demand/2。 在 init/1 中，設置了初始狀態，就像在 GenServers 中做的那樣，但更重要的是，我們將自己標記為生產者。 init/1 函數的回傳值是 GenStage 對處理程序 (process) 進行分類的依據。

handle_demand/2 函數是生產者被定義的主要部份。 它必須由所有 GenStage 生產者實現。 在這裡，我們回傳消費者所需的一組數字，並累加我們的計數器（counter）。 來自消費者的需求，也就是上面程式碼中的 demand，被表示為一個能夠處理事件量的相對應整數；預設為 1000。

生產者 消費者 (Producer Consumer)

現在已經有了能產出數字的生產者，接著看生產者－消費者。 我們希望從生產者索取數字，濾除奇數，並能回應需求。

touch lib/genstage_example/producer_consumer.ex

現在更新檔案，使其看起來像範例內的程式碼：

defmodule GenstageExample.ProducerConsumer do
  use GenStage

  require Integer

  def start_link(_initial) do
    GenStage.start_link(__MODULE__, :state_doesnt_matter, name: __MODULE__)
  end

  def init(state) do
    {:producer_consumer, state, subscribe_to: [GenstageExample.Producer]}
  end

  def handle_events(events, _from, state) do
    numbers =
      events
      |> Enum.filter(&Integer.is_even/1)

    {:noreply, numbers, state}
  end
end

你可能已經注意到，在生產者-消費者裡，init/1 已經引入了一個新選項和一個新函數： handle_events/3。 通過 subscribe_to 選項，指示 GenStage 讓我們在特定生產者的對話中（communication）。

handle_events/3 函數是我們的主力，接收傳入事件、處理 (process) 它們並回傳轉換後的集合。 而消費者的實現方式也大致相同，但重要的區別在於 handle_events/3 函數的回傳值以及如何被使用的。 當將處理程序標記為 producer_consumer 時，在範例中是 tuple 的第二個引數 - numbers 是用於達到下游的消費者需求； 但在消費者中，這個值會被丟棄。

消費者

最後但並非最不重要的是消費者。 開始囉：

touch lib/genstage_example/consumer.ex

由於消費者和生產者-消費者如此相似，所以程式碼看起來不會有太大的不同：

defmodule GenstageExample.Consumer do
  use GenStage

  def start_link(_initial) do
    GenStage.start_link(__MODULE__, :state_doesnt_matter)
  end

  def init(state) do
    {:consumer, state, subscribe_to: [GenstageExample.ProducerConsumer]}
  end

  def handle_events(events, _from, state) do
    for event <- events do
      IO.inspect({self(), event, state})
    end

    # As a consumer we never emit events
    {:noreply, [], state}
  end
end

正如在前面章節中所介紹的，消費者不會產出事件，所以 tuple 中的第二個值將被丟棄。

將所有角色組合在一起

現在已經建立了生產者、生產者-消費者和消費者，我們已經準備好將所有東西串在一起了。

首先打開 lib/genstage_example/application.ex 並將新處理程序加入到 supervisor 樹：

def start(_type, _args) do
  import Supervisor.Spec, warn: false

  children = [
    {GenstageExample.Producer, 0},
    {GenstageExample.ProducerConsumer, []},
    {GenstageExample.Consumer, []}
  ]

  opts = [strategy: :one_for_one, name: GenstageExample.Supervisor]
  Supervisor.start_link(children, opts)
end

如果一切都是正確的，專案就可以執行，應該會看到所有東西都能正常運作：

$ mix run --no-halt
{#PID<0.109.0>, 2, :state_doesnt_matter}
{#PID<0.109.0>, 4, :state_doesnt_matter}
{#PID<0.109.0>, 6, :state_doesnt_matter}
...
{#PID<0.109.0>, 229062, :state_doesnt_matter}
{#PID<0.109.0>, 229064, :state_doesnt_matter}
{#PID<0.109.0>, 229066, :state_doesnt_matter}

我們做到了！正如預期那樣，應用程式只會產出偶數，而且它做的非常 快。

在此刻，我們有了一個工作管線（pipeline）。 生產者產出數字；生產者-消費者丟棄奇數；消費者則顯示以上這些並繼續流程。

多個生產者或消費者

在簡介中提到，有一個以上的生產者或消費者是可行的， 現在來看看這個。

如果檢查範例中的 IO.inspect/1 輸出，會看到每個事件都單獨由一個 PID 處理。 現在通過修改 lib/genstage_example/application.ex 對多個 worker 進行一些調整：

children = [
  {GenstageExample.Producer, 0},
  {GenstageExample.ProducerConsumer, []},
  %{
    id: 1,
    start: {GenstageExample.Consumer, :start_link, [[]]}
  },
  %{
    id: 2,
    start: {GenstageExample.Consumer, :start_link, [[]]}
  },
]

現在已經配置了兩個消費者，看看如果執行應用程式會得到什麼：

$ mix run --no-halt
{#PID<0.120.0>, 2, :state_doesnt_matter}
{#PID<0.121.0>, 4, :state_doesnt_matter}
{#PID<0.120.0>, 6, :state_doesnt_matter}
{#PID<0.120.0>, 8, :state_doesnt_matter}
...
{#PID<0.120.0>, 86478, :state_doesnt_matter}
{#PID<0.121.0>, 87338, :state_doesnt_matter}
{#PID<0.120.0>, 86480, :state_doesnt_matter}
{#PID<0.120.0>, 86482, :state_doesnt_matter}

正如所看到的，現在擁有多個 PID，只需簡單的加入一行程式碼並提供消費者 ID 即可。

使用案例

現在已經介紹了 GenStage 並建立了第一個範例應用程式，但 GenStage 有哪些 真實 案例？

而這些還只是 GenStage 的 一小部分 可能性。

Caught a mistake or want to contribute to the lesson? Edit this lesson on GitHub!
Previous Lesson

StreamData

基於案例（example-based）的單元測試函式庫，例如 ExUnit，是個極佳的工具來協助驗證程式碼是否以如你預期的方式工作。 但是，基於案例的單元測試有一些缺點：

  • 由於並非測試所有輸入情境，因此容易遺漏邊界案例。
  • 可以編寫這些測試案例而無需仔細考慮需求。
  • 想以多個案例測試單一函數時，這些測試可能會非常冗長。

在本課程中，將探討 StreamData 如何幫助我們克服上述一些缺點

Next Lesson

基礎

Ecto 是一個官方 Elixir 專案，提供資料庫封裝 (wrapper) 和整合查詢語言。通過 Ecto，能夠建立遷移 (migration)、定義結構描述 (schema)、插入 (insert) 和更新記錄 (update) 並查詢 (query)。