Wie mache ich einen GenServer, der Nachrichten mit einer bestimmten Rate verarbeitet? (alle n Sekunden)

Question

Einer meiner Dienste spricht mit einer externen API, die Rate-limited ist, also möchte ich sicherstellen, dass ich nicht mehr als 1 Anruf alle 10 Sekunden sende.

Mein naiver Ansatz wäre, einen lang andauernde API-Service zu haben, und es ist Zeit nach jedem Anruf aus:

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

Ich bin nicht sicher, ob eine richtige Art und Weise gibt es, dies zu tun.

Answer 1

Bearbeiten: Die ursprüngliche Lösung löschte Nachrichten zwischen 10s Ticks, wie burmajam vorgeschlagen. Die Bearbeitung bietet eine geeignetere Lösung.

---

EDIT

Aufgrund der Tatsache, dass GenServer handle_ * Funktionen tatsächlich keine Nachrichten aus der Warteschlange empfangen, sondern sie einfach verarbeiten, können wir nicht ausnutzen Muster selektiv nur jeweils 10s von der erhalten passend Prozesse Warteschlange.

Da wir Nachrichten in der Reihenfolge ihrer Ankunft abholen, müssen wir daher die interne Warteschlange als Teil des Status von GenServer verwenden.

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 

So sind wir nur Nachrichten von Prozess-Warteschlange Zustand Warteschlange bewegt und wir verarbeiten, um den Kopf, wenn wir uns signalisieren, dass 10s bestanden haben.

Aber am Ende erreichen wir tatsächlich das gleiche Ergebnis, als wenn wir den Prozess für 10 Sekunden schlafen legen würden. Ihre Lösung scheint einfacher und erzielt das gleiche Ergebnis.

---

Die Lösung auf How to run some code every few hours in Phoenix framework?

Zuerst lassen Sie Ihre GenServer store eine Fahne in seinem Zustand (Arbeit = true/false) basiert.

Dann lassen Sie GenServer Process.send_after verwenden, um sich selbst zu signalisieren, wenn es funktionieren kann. Sie empfangen das Signal in handle_info, in dem Sie das work Statusflag zu true setzen.

Jetzt beachten Mustererkennung auf Zustand in der handle_cast Funktion: Es wird nur die Nachricht abholen, wenn work State Flag gleich wahr ist. Andernfalls werden die Nachrichten in die Warteschlange gestellt.

Und nachdem Sie die Nachricht an externem Service senden, führen Sie Process.send_after wieder das nächste Signal zu planen und Zustand zurück, die work Flags auf false gesetzt hat, um ab dem nächste Nachricht zu verhindern, dass sofort abgeholt.

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