Search⌘ K
AI Features
Home
Courses
Building Real-time Applications with Phoenix & Elixir

Adding Concurrency and Channels

Explore how to implement concurrency in real-time applications using GenStage's ConsumerSupervisor in Elixir. Learn to dynamically spawn worker processes that handle parallel data streams, integrate Phoenix channels to push messages to users, and configure concurrency using max_demand settings. This lesson helps you build a scalable data pipeline and deliver real-time updates efficiently.

We'll cover the following...

Concurrency

A scalable data pipeline must handle multiple items simultaneously; it must be concurrent. GenStage has a solution for adding concurrency to our pipeline with the ConsumerSupervisor module. This module allows us to focus on defining the pipeline and letting the library take care of how the concurrency will be managed.

ConsumerSupervisor is a type of GenStage consumer that spawns a child process for each item received. The concurrency is controlled via setup options, and it behaves exactly like a consumer. Every item spawns a new process; they’re not re-used, but this is cheap to do in Elixir.

Note: We make our system concurrent by creating processes that work. The BEAM then makes that system parallel by taking the concurrent work and running it over multiple cores simultaneously. How parallel execution happens is entirely handled by the BEAM.

Our final result in this section will look like this:

Our Consumer has been replaced by a ConsumerSupervisor, which can dynamically spawn worker processes. Let’s walk through adding ConsumerSupervisor to our pipeline.

Elixir
# hello_sockets/lib/hello_sockets/pipeline/consumer_supervisor.ex
defmodule HelloSockets.Pipeline.ConsumerSupervisor do
  use ConsumerSupervisor
  alias HelloSockets.Pipeline.{Producer, Worker}
  def start_link(opts) do
    ConsumerSupervisor.start_link(__MODULE__, opts)
  end
  def init(opts) do
    subscribe_to = Keyword.get(opts, :subscribe_to, Producer)
    supervisor_opts = [strategy: :one_for_one, subscribe_to: subscribe_to]
    children = [
      %{id: Worker, start: {Worker, :start_link, []}, restart: :transient}
    ]
    ConsumerSupervisor.init(children, supervisor_opts)
  end
end

This ConsumerSupervisor module is, fittingly, a mix of common Supervisor and Consumer ...