Gain insights into data processing and concurrency in Elixir. Explore GenServer, GenStage, Flow, and Broadway to enhance performance, set up pipelines, and handle increased workloads efficiently.

Concurrent Data Processing in Elxir by The Pragmatic Programmer-01.png

myDokcer.tar.gz

ch1_1

ch_3

ch_3.1

ch1_1-task supervisor

ch_4.1

ch_5.1 RabbitMQ

Ch_5.2 RabbitMQ

Most engineers don’t think that concurrency is distinct from programming. Data processing includes transforming information in some way, like reporting, data aggregation, or analytics. On the other hand, concurrency involves significantly improving performance.

This course is designed to help you navigate the world of concurrency tools available in the Elixir ecosystem. You’ll explore popular modules and libraries like GenServer, GenStage, Flow, and Broadway. You’ll be able to ingest data, set up pipelines, and run concurrent processes with Elixir.

By the end of the course, you’ll have a wide range of new techniques to simplify your product, improve code performance, and make your application more resilient to increased workloads.

Concurrent Data Processing in Elixir

# Introduction to dispatchers

There is one important component of `GenStage` we haven’t talked about yet—the **dispatcher**. When `:producer` and `:producer_consumer` stages send events to consumers, it is the dispatcher that sends the events. So far, we’ve used the default `DemandDispatcher`, but `GenStage` comes with two more. Let’s see what they do and how we can use them.

# `DemandDispatcher`

We can specify the kind of dispatcher to use when we initialize the process using the `:dispatcher` key. The default is **`DemandDispatcher`**, which is equivalent to this configuration:

def init(state) do
 {:producer, state, dispatcher: GenStage.DemandDispatcher}
end

The `DemandDispatcher` dispatcher sends events to consumers with the highest demand first. It is the dispatcher that we'll use most often. However, there are cases where we may want to route events to consumers using a different strategy, where `BroadcastDispatcher` and `PartitionDispatcher` come in. Let’s look at `BroadcastDispatcher` first.

# `BroadcastDispatcher`

We can switch to `BroadcastDispatcher` by setting `:dispatcher` to `GenStage.BroadcastDispatcher` in our `init/1` function:

```elixir
{:producer, state, dispatcher: GenStage.BroadcastDispatcher}
```
As its name suggests, `BroadcastDispatcher` sends the events supplied by the `:producer` or `:producer_consumer` to all consumers subscribed to it. This is useful when we have different types of consumers subscribed to the same producer and need the same data but for other purposes.

When `BroadcastDispatcher` is used, consumer stages can filter the events they receive. This means that each consumer can opt for specific events and discard the rest. All we have to do is use the `:selector` setting when subscribing to the producer as shown below:

```elixir
def init(state) do
  selector =
     fn incoming_event ->
       # we can use the event to decide whether
       # to return `true` and accept it, or `false` to reject it.
      end
  sub_opts = [
    {SomeProducer, selector: selector}
  ]
  {:consumer, state, subscribe_to: sub_opts}
end
```

As we can see, the `:selector` is just a function, which we can optionally pass to each producer individually. The consumer has to examine each message and decide whether they want it or not.

The other type of dispatcher we’ll going to look at also helps us route events to specific consumers, but it works in a slightly different way.

# `PartitionDispatcher`


Unlike `BroadcastDispatcher`, where the consumer has to check each event, `PartitionDispatcher` leaves this responsibility to the producer. The producer examines each event and assigns it to a **partition**. We can think of partitions as buckets in which we put one or more events. Based on that partition, consumers can then tell the producer the type of events they want to receive.

There are two extra arguments we need to pass when configuring `PartitionDispatcher`—`:partitions` and `:hash`. Here is an example:

```elixir
def init(state) do hash =
fn event ->
#we can use the event to decide which partition 
# to assign it to, or use `:none` to ignore it. {event, :c}
end
opts = [
partitions: [:a, :b, :c], hash: hash
]
{:producer, state, dispatcher: {GenStage.ParitionDispatcher, opts}}
end
```

The `hash` function is similar to the `selector` function in `BroadcastDispatcher`, but it has to return a tuple rather than a boolean. The tuple should contain the event we want to dispatch and the name of the partition. Partitions are chosen by us and could be any Elixir term. Here, we pick `:a`, `:b`, and `:c`. We can also ignore an event by passing `:none`, which we don’t have to define in our `:partitions` list. Also, notice that the `:dispatcher` value is now a tuple. This allows us to pass the dispatcher configuration options as the second element in the tuple.

Now that the producer is configured, consumers can subscribe to one of the partitions when initializing:

```elixir
sub_opts = [
  {SomeProducer, partition: :b}
]
{:consumer, state, subscribe_to: sub_opts}
```

# Create our dispatcher

`BroadcastDispatcher` and `PartitionDispatcher` are a great addition to the default `DemandDispatcher` and should help us adapt our data processing pipeline to an ever wider variety of use cases.

However, if they still don’t match what we're trying to accomplish, we can also create our dispatcher from scratch by implementing the `GenStage.Dispatcher` behaviour. Check out the `Dispatcher` module's [documentation](https://hexdocs.pm/gen_stage/GenStage.Dispatcher.html#content) for more information on what callbacks we have to implement.

___

Learn how GenStage uses different dispatchers to send events to consumers.

Getting Started

Easy Concurrency with the Task Module

Long-Running Processes Using GenServer

Data Processing Pipelines with GenStage

Process Collections with Flow

Data Ingestion Pipelines with Broadway

Concluding the Course

Choose the Right Dispatcher

Introduction to dispatchers