elixir - 如何制作一个以特定速率处理消息的 GenServer？ (每n秒)

Question

我的一项服务与速率受限的外部 API 进行通信，因此我想确保每 10 秒发送的调用不超过 1 个。

我的天真的方法是拥有一个长期运行的 API 服务，并在每次调用后超时:

def handle_cast({:call_api, data}, state) do
  send_to_external_api(data)
  :timer.sleep(10000)
  {:noreply, state}
end

我不确定是否有正确的方法来做到这一点。

Answer 1

编辑:原始解决方案在 10 秒内丢弃消息，如 burmajam建议。编辑提供了更合适的解决方案。

---

编辑

由于 GenServer handle_* 函数实际上并不从队列接收消息，而只是处理它们，因此我们无法利用模式匹配来选择性地从进程队列中每 10 秒接收一次消息。

因此，由于我们按照消息到达的顺序来获取消息，因此我们需要内部队列作为 GenServer 状态的一部分。

defmodule RateLimited do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, %{queue: []})
  end

  def init(state) do
    allow_work()
    {:ok, state}
  end

  def handle_cast({:call_api, data}, %{"queue" => queue} = state) do
    {:noreply, %{state | queue: queue ++ [data]}}
  end

  def handle_info(:work, %{"queue" => [data | queue]} = state) do
      send_to_external_api(data)
    allow_work()

    {:noreply, %{state | queue: queue}}
  end

  defp allow_work() do
    Process.send_after(self(), :work, 10000) # after 10s
  end

  defp send_to_external_api (data) do end
end

因此，我们只是将消息从进程队列移动到状态队列，并在我们向自己发出 10 秒已经过去的信号时处理头部。

但最终，我们实际上获得了与让进程休眠 10 秒相同的结果。您的解决方案似乎更简单并达到相同的结果。

---

解决方案基于How to run some code every few hours in Phoenix framework?

首先，让您的 GenServer 在其状态中存储一个标志(work = true/false)。

然后让 GenServer 使用 Process.send_after 向自己发出信号，告诉自己何时可以工作。您在 handle_info 中收到信号，将 work 状态标志设置为 true。

现在请注意 handle_cast 函数中状态的模式匹配:只有当 work 状态标志等于 true 时，它​​才会拾取消息。否则，消息将被放入队列中等待。

将消息发送到外部服务后，再次运行 Process.send_after 来安排下一个信号并返回将 work 标志设置为 的状态false 以防止立即拾取下一条消息。

defmodule RateLimited do
  use GenServer

  def start_link do
    GenServer.start_link(__MODULE__, %{work: false})
  end

  def init(state) do
    allow_work()
    {:ok, state}
  end

  def handle_cast({:call_api, data}, %{"work" => true} = state) do
    send_to_external_api(data)
    allow_work()
    {:noreply, %{state | work = false}}
  end

  def handle_info(:work, state) do
    {:noreply, %{state | work = true}}
  end

  defp allow_work() do
    Process.send_after(self(), :work, 10000) # after 10s
  end
end