Forem: Mateus Viegas

Messaging: an Introduction

Mateus Viegas — Sun, 21 Feb 2021 18:16:00 +0000

tl/dr

This post is the first one of a series about basic concepts within distributed architectures.

1) Messaging: An introduction

2) Messaging: Practical basic concepts with System.Threading.Channels

3) Messaging: Interaction Styles

4) Messaging: Consistency and Resiliency Patterns

The goal of this series is to provide new venturers on distributed systems world with basic concepts of messaging, the tradeoffs of this kind of resource and how to provide solutions for some of these tradeoffs in order to achieve the most important benefit of messaging: loose coupling and high scalability within a distributed architecture.

Messaging is one of the pillars for this kind of architecture to be sustainable. Here we will showcase basic theoretical concepts such as Messages and their types, as well as Transport Channels and their types. Also, we will give a high level overview about when you should or shouldn't consider messaging.

Loose coupling, but why?

First things first: we can define messaging as a way of communication between two or more resources. In this manner, these resources communicate by sending and receiving messages through a messaging infrastructure instead of doing direct and synchronous requests to each other.

This is achieved by the exchange of messages via one or more message channels: a sender, which is some component within a system, writes messages to a channel inside this messaging infrastructure, and a receiver, another component, reads messages coming in via this channel. Thus, when messaging is introduced and sync communication is removed, we also remove direct coupling between these components. We can say then that these components are loosely coupled.

As pretty as it might seems, one question remains though: why would we want to make components within a system loosely coupled? To answer that question, let's understand one specific type of coupling that might be the greatest reason for using messaging and distributing a system.

Temporal coupling

An application usually has some IO-bounded dependencies such as disks and networks. These dependencies might be databases, calls to external APIs, email sender services, file storages, or even some other external system, etc. Each time a request cross an IO boundary, it will have its response coupled to the result/response of this initiated IO operation.

Even with modern concepts such as Task-based programming in C#, the so called asynchronous code is, de facto, async only from a thread perspective. This means, that applications like a web API, will still have the incoming requests waiting for each one of these async blocks of code to complete in order respond the client. The asynchronicity mentioned here relies on how these requests are executed from a thread perspective, meaning they are async because they do not lock the calling thread, a CPU resource, while IO operations are in place. What happens is that the application, when calling an async code, passes the responsibility of the execution continuation to another thread and frees up the calling one. This is actually really important, since there's a limited number of threads to execute CPU-bounded operations on the thread pool. Imagine the following situation:

There's an incoming request;
This request does a query to the database and return a response with the result;
- Since database queries are IO operations, the calling thread might be blocked while awaiting for the database to execute the query. In this case, imagine that there are multiple requests coming in and that the database is, by some reason, overloaded and not responding properly. Since the number of CPU-bounded threads is limited, there might come a time when there's no more free threads to handle incoming requests: starvation happens. In this case, the application might be unresponsive or with a very high response times for new requests, since there are no more free threads to execute and it has to wait more threads to be available.
- Now, imagine that these incoming requests access the database via an async method that executes the query. When that happens, a state machine is created with the request context and the control of this state machine is passed to a new thread, so that the CPU-bounded calling thread gets free to be used by any other incoming request. This another IO-bounded thread executes its operation on the database and, since it has this state machine, it is able to provide a continuation to the request through a synchronisation context, a.k.a., returning a response to the API with the query result. In this case, we give some buffer for our application performance, allowing our thread pool to have the incoming threads relying on CPU-bounded work only and being released more frequently to handle new requests.

Tl/dr: this whole explanation was to clarify that executions with the async keyword are, from a temporal perspective, still synchronous. The overall response time still depends on the database execution. Or, perhaps, on another external and unstable web API/system. And that's why this type of coupling is called temporal coupling and might be an issue if you need an application to have fast response times.

Messaging allows us to remove temporal coupling by deferring requests to be executed either in background or by third-party applications. In this case, a system component sends a message containing some sort of data through a channel and another system component, which is connected to this channel, receives this message and execute. Therefore, the component that sends the message is temporally coupled only to the message dispatching and not to an entire ecosystem of dependencies that there might exist.

Message Types

Messages are chunks of data structured with both a body and also some headers. The body contains the central information and the message payload, and the headers might have some really useful metadata such as an unique message identifier, one or more correlation identifiers that could be used to correlate operations on different systems, etc. What's important to understand here is: messages are literally DTOs (data transfer objects). These DTOs are then serialised and sent via a channel. Either in JSON, binary, XML or any other format.

It is also common to think about messages from a semantic point of view, giving them meaning. This allows them to be categorised into types. Some of the most popular message types are commands and events.

Commands

Commands can be understood as requests specifications. A descriptive method name with zero or more parameters. Imagine, for instance, a CreateUserCommand. This command might have some parameters such as Email and Password. We could describe this message as the following class:

public class CreateUserCommand
{
   public string Email { get; set; }
   public string Password { get; set; }
}

When this class is sent over a channel, the recipient system would then read this specification and execute the operation. The meaning of a command is basically "do something with these attributes", in this case, this message is telling to the recipient: "create a user with this email and this password".

Events

On the other hand, we have events, that indicates and describe something that happened within the system. So for a CreateUserCommand, we could have a UserCreatedEvent describing a side-effect on our system, result from a previous operation. In this particular case, the event would be pretty similar to the command.

public class UserCreatedEvent
{
   public string Email { get; set; }
}

You could read this as "an user was created with this email". This information could then be broadcasted via messaging, so that any other interested part of the system could receive this information and do something about it, like sending a welcome email, for instance.

Channels

The Enterprise Integration Patterns book has the following description about channels: "when an application has information to communicate, it doesn't just fling the information into the messaging system, it adds the information to a particular Message Channel. An application receiving information doesn't just pick it up at random from the messaging system; it retrieves the information from a particular Message Channel."

There are mainly two types of message channels:

Point-to-point: this type of channel delivers the message to exactly one, and only one, consumer that is listening to the channel. If there are multiple consumers, it means that they will concur for the incoming messages and each message will only be read by one consumer. To have multiple and competing consumers in a point-to-point channel could be useful to increase the throughput of the channel, if we have a lot of messages being sent to it, and is basically a way of scaling the message consumption. Commands are often sent through this type of channel.
Publish-subscribe: this type of channel might have multiple consumers that receive, each one, a copy of the message. This type of channel is particularly good for some scenarios. With pub/sub we can easily move information around many interested parties with no temporal coupling between them. One system X can know what happens to other system Y by receiving messages sent by Y to a message channel that X, by its turn, is listening to. That way, we can easily move information around many components within a system with no tight coupling between them, making the integration process between these components smoother. Events are often sent through this type of channel.

First trade-offs

Even though messaging is often referred as a way of integrating and remove coupling between distributed systems, we can use the same concept to extend functionality inside one single system. By using the concept of commands and events within an application, we can easily expand its features without changing existing ones. I talk a little bit more about it in an old post about the correlation between "the open-closed principle and event-driven architectures".

However, just as anything regarding to software engineering and architecture, messaging has its trade-offs. By delegating the message handling to an external consumer via a message channel we introduce a single point of failure: the messaging infrastructure. A single point of failure, commonly named SPOF, it is a component within a system that, if fails, will stop the entire system of working.

Therefore, since all the communication between the systems rely on a central piece of communication, the messaging infrastructure, it becomes a requirement that this piece has a very high level of availability.

There are still many other trade-offs of applying messaging and distributing your system, such as maintenance, monitoring and development complexity introduced by many side-effects caused by the decoupling, and we will talk more in detail about them on the next posts of this series.

Wrap-up

Messages are a nice concept and are very fundamental when we talk about distributed systems and architectures. The concept itself is an abstraction, which implementations are diverse, as well as the possible infrastructure resources to handle it.

In this introductory post, we showcased how we exchange temporal for loose coupling with messaging, and why it might be good. We also introduced basic message types, channel types and also discussed briefly about the consequences of introducing messaging to a system.

On the next post will showcase an abstraction of the concepts showcased here: we will remove temporal coupling in a web API by sending messages via channels created with C# and the System.Threading.Channels library.

A brief brainstorm about the open-closed principle and event-driven architectures

Mateus Viegas — Thu, 04 Jun 2020 19:38:12 +0000

tl/dr

The open-closed principle, one of the SOLID ones, states that one system/module/program should be "open for extension, but closed for modification." I've always found the statement itself quite vague. In this post, we are going to discuss in a little more practical way what it might mean from a practical point of view, both in code and architecture perspectives. We're going to talk about some specific concerns here, such as Domain-Driven Design, Pub/Sub and Message Brokers, so it might be useful to at least make yourself familiar with these concepts first.

Giving meaning to what it seems obvious

The first question that came to my mind as I was first reading about the OCP was: how can we make a system/module/functionality open for extensions but closed to modifications? And why should we do it?

We can go from micro to macro in order to answer the question, so I will start with a classic: an oriented-object example.

Let's think about a chatbot, for instance. There's one simple basic and starter thing that it must be able to do that is: to talk. And, as for every conversation, a mean must be provided between the two parties so that they can listen to one another and understand each other. If our chatbot is built around the HowsEverything messenger platform, we might have something like:

public class Chatbot
{
    void Answer(Message message)
   { 
      // Some specific implementation about to deal with the HowsEverything platform
   }
}

With the above example, both a language and a channel are specified so that this conversation is stablished through the HowsEverything app. But imagine that someday in the future, my chatbot is required to expand to some other platforms like Letter or Nosekindle. As the logic is platform-specific, we would have to change its functionality in order to embrace this new requirement and, therefore, break the open-closed principle. But that's not what we want.

We can take a further step and make our chatbot platform-agnostic, so that it can be extensible to which platform it may arise as a new requirement. One possibility to do so is to introduce some abstraction to our code:

public abstract class Chatbot
{
    void abstract Answer(Message message);
}

public class HowsEverythingChatbot
{
    void Answer(Message message) { // specific implementation }
}

public class NosekindleChatbot
{
    void Answer(Message message) { // specific implementation }
}

Doing that, our user might be able to always communicate with our chatbot in despite of the chosen platform. And we are also able to expand our chatbot from being platform-specific to being platform-agnostic.

Abstractions are a beautiful way to achieve the open-closed principle by code, but that was just an example of this principle from a code perspective for you to be familiar with it. Now let's jump to the architecture level.

Extensibility with an Event-Driven Architecture

To think about design and architecture is a exercise that I encourage every developer to do. Whether to think about it up-front or embrace the evolutionary side of things (I know, both ways could and should work together, but there's still a lot of discussion about it). The fact is: a successful system will grow over time. If it will be to 10, 100, 1k, 10k or 100k users depends on its purpose. But it will grow, and you'll have to make sure that it's growing wealth. Therefore, one approach that I really like and to think about in large scale systems is an event-driven architecture. And the first things that I like to start thinking when I have to decided whether or not to follow this approach is to learn about my requirements and see if they fit as a rich domain with its commands and events.

We can think about those like this: every action has a consequence. We can thrive this axiom to the fact the every user action within a system has a result. Improving the language: every successful command generates an event. An UserSignUpCommand, where a new user is created, can be successfully stated as an UserSignedUpEvent. Note this event is an confirmation that an intention has occurred in a correct manner. And what does the commander do with that information? It simply tells someone that it is there, i.e., a command raises an event so that interested parties may listen to this event and do whatever they'd like to do with that piece of knowledge.

Well, now imagine that the first requirement about the above example was simple: "we must notify the user that he his registration was successful". So after a command has been processed, we listen to an event and handle it so that, in this handle, we send an email notification. Looks quite reactive, and it is, but quite simple still, right?

Now imagine that some more things happen:

The marketing people should know about a new customers so they can start to track them in a customer funnel.
The billing people should know about new customers, so that they can give him a free gift, like a 1-month subscription voucher.
And so on ...

And now is where this kind of approach shines: in an event-driven architecture we can have as many stakeholders as needed for a single event. That means, that each new part that desires to do a new thing with this information must simply subscribe to an event, in order to handle it. And by doing so, we do not change a line, neither in our command nor in our first handler. Congratulations: you've just extended your system without moving a single character of code of one existing and working functionality.

But, how can we after a successful command, notify other parties about an event? Well, one approach is to follow the pub/sub pattern, where the application must simply publish this event to someone who's responsible to handle it to the interested parties, so that these events are correctly delivered and handled by these parties. It can be done in a bunch of different ways, but before deciding how we must decide when:

Synchronously, with an in-memory mediator.
Asynchronously, with an event-bus built around a message broker.

And there's a lot of room to talk about this kind of coupling (don't fool yourself, async processes are still coupled, but to time instead of space). But that's not the goal of this post.

Finally, with this approach, we accomplish exact what the open-closed principle states: we extend our system, enriching its behaviour, without modifying what's already there. And that's what this principle is about.

Going further with Domain-Driven Design

Domain and event-driven systems share a lot of similarities and we can use both approaches to, not only better develop one specific domain, but also to communicate specific domain actions with N aggregate roots in the same application. In case you want to go more in-depth, Microsoft has a really good article with even more references about the theme.

For instance, let's take a given aggregate root such as Order in an e-commerce system. A user can search goods, add them to cart and execute the checkout with a PlaceOrderCommand. This command, by its turn raises an OrderCreatedEvent. At last, this event might be handled by different subdomains to performs tasks related to:

payments
warehouse, supply chain and logistics
order tracking by the user

As I first learned about this, my eyes shined and I thought that I was entering a whole new world. But of course, once that with great power comes great responsibility, one must also be aware that these approaches have a plenty of drawbacks, specially if the event distribution is made asynchronously. In such case, you'll probably have to dive into some patterns to deal with the distribution of events such as outbox pattern and event sourcing. Beyond that, it is possible that you'll have to deal with distributed transactions and sagas.

Outcome

The open-closed principle is one of the most important principles regarding the growth of a system. To achieve it through event-driven architectures looks like an almost perfect match, as it allows new functionalities to be added with no disruption of old ones as the domain expands. And by applying event-driven we can automatically apply the open-closed principle to the architectural level.

However, one should be aware that this approach carries with it a lot of advanced requirements and complexities, so one should really outweigh the pros and cons in order to decide if it fits its use case.

What do you think about this theme? Let me know in the comments!

Explorando System.Threading.Channels

Mateus Viegas — Sun, 17 May 2020 20:57:26 +0000

tl/dr

Este post tem como objetivo ser uma explicação prática da API System.Threading.Channels e como ela pode ser útil para implementar comunicação entre Tasks seguindo padrões pub/sub. Foi inicialmente publicado no meu blog 🚀.

Produtores x Consumidores

Neste post do blog da Microsoft é levantada a motivação pra criação da lib System.Threading.Channels. São recorrentes, até mesmo no dia a dia, os cenários de se produzir algo por uma ou mais partes e que esse algo vai posteriormente ser consumido por uma ou mais outras partes. E ainda que a produção e o consumo sejam feito de maneiras independentes por cada uma dessas partes, ambas precisam estar conectadas uma(s) à(s) outra(s) para que isso aconteça. Por exemplo, uma encomenda enviada por um remetente, para ser entregue a um destinatário, precisa passar por todo um processo logístico até que isso aconteça. Esse processo pode ser abstraído como um canal entre as duas partes interessadas, e esse canal que permite o recebimento.

É justamente uma abstração dessa conexão, entitulada de channel que é criada com a lib System.Threading.Channels. Para usá-la, basta instalar um package nuget, que é compatível com o .NET Standard(>= 1.3), .NET Core (>= 3.1) e até mesmo com o NET Framework (>= 4.6).

Dissecando um channel

Abrindo channels

Um channel pode ser visto, portanto, como uma estrutura de dados básica pra estabelecer uma conexão entre produtores e consumidores. É uma estrutura genérica e que pode ser criada através da classe System.Threading.Channels.Channel<T> pelos seguintes métodos, que são static factories:

CreateBounded<T>(int capacity): cria um bounded channel, que é um canal que tem uma capacidade de objetos (mensagens) definida. Se um channel é criado desta forma, significa que, uma vez que ele atingiu sua capacidade máxima, ele só poderá receber um novo objeto depois que outro objeto previamente colocado nele por um writer seja consumido por um reader.
Channel<T> CreateBounded<T>(BoundedChannelOptions options): tem exatamente o mesmo comportamento da factory acima, com o adicional da possibilidade de definição de opções do canal. Na data de escrita deste post, essas opções setam não só a capacidade, mas também como o channel vai se comportar no caso de ter chegado no topo de sua capacidade. Isso é definido através da propriedade FullMode, um enumerado com quatro opções:
- Wait: espera que o canal tenha espaço livre para completar uma operação de escrita.
- DropNewest: descarta o item mais novo já escrito mas ainda não lido no canal, para abrir espaço para o que acabou de chegar.
- DropOldest: de maneira semalhante à operação acima, porém com o item mais antigo já escrito mas ainda não lido no canal.
- DropWrite: descarta outro item que esteja sendo escrito no canal.
Channel<T> CreateUnbounded<T>(): cria um unbounded channel - um canal sem capacidade definida, aonde o céu (na verdade, a quantidade de memória RAM disponível) é o limite. Segundo testes de benchmark apresentados por este blog, unbounded channels são mais performáticos. Contudo, criar canais dessa forma pode ser perigoso uma vez que eles não tem limitação e podem consumir toda a memória disponível. Por esta razão, é dito que eles são menos seguros que os seus irmãos apresentados anteriormente. Deve ser analisado o tradeoff de segurança x performance, caso o nível de requisitos seja crítico a este ponto. Caso seja possível garantir a capacidade máxima de maneira externa, pode valer o risco.
Channel<T> CreateUnbounded<T>(UnboundedChannelOptions options): tem exatamente o mesmo comportamento da factory acima, com o adicional de definição de opções do canal. E esta classe tem uma propriedade que é bem interessante:
- AllowSynchronousContinuations: de acordo com a doc oficial, essa propriedade tem uma comportamento bem interessante. Ela permite, de certaforma, que um produtor vire, temporariamente, um consumidor. Como? Vamos entrar nesse detalhe mais abaixo após entender como produtores (writers) e consumidores (readers) se comportam.

Ambas as classes de opções de unbounded e bounded channels herdam de ChannelOptions que, por sua vez, possui duas propriedades:

SingleWriter: tem como default false. Se setada como true, é garantida que acontecerá apenas uma operação de escrita por vez. Por padrão, não existe essa constraint, e a própria estrutura da api de channels pode otimizar algumas operações de escrita internamente.
SingleReader: o mesmo comportamento da operação acima, porém com operações de leitura no canal.

No exemplo abaixo, criamos um bounded channel que receberá objetos do tipo int com capacidade máxima de 1.

var channel = System.Threading.Channels.Channel.CreateBounded<int>(new BoundedChannelOptions(1)
{
    FullMode = BoundedChannelFullMode.Wait
});

Encerrando channels

Um canal pode ser encerrado através de um recurso chamado completion. Este recurso é controlado sempre pelos produtores (writers) e pode ser ativado através dos métodos Complete e TryComplete, que podem inclusive receber uma exceção para indicar que o channel foi encerrado devido à um erro.

Durante uma operação de leitura, os consumidores (readers) podem, por sua vez, verificar se um canal foi encerrado chamando channel.Completion.IsCompleted.

Produzindo e consumindo mensagens

Mensagens podem ser produzidas no canal através de um Writer e lidas por um Reader, propriedades da classe Channel<T>. Isso pode através dos seguintes métodos:

1) TryWrite/TryRead: tentam escrever/ler uma mensagem de maneira síncrona no canal. Caso não seja possível, pelas opções definidias, a mensagem é descartada e o método retorna false. Caso contrário, retorna true. No exemplo abaixo, escrevemos no console se uma mensagem não pôde ser escrita no canal. Isso ficará mais claro no exemplo inteiro que daremos um pouco mais à frente, aonde colocamos um delay entre a leitura e a escrita pra simular cenários em que o canal esteja funcionando na capacidade total e, por consequência e a depender de como foi configurado, esteja impedido de receber novas escritas de mensagens.

for (int i = 0; ; i++)
{
    if (!channel.Writer.TryWrite(i))
    {
        Console.WriteLine($"Dropping {i}");
    }

    await Task.Delay(1000);
}

2) WaitToWriteAsync/WaitToReadAsync: retornam uma ValueTask<bool> e indicam se o canal está disponível para escrever/ler mensagens, baseados nas opções e capacidade definidas. Por padrão, estes métodos não dispararão uma exceção caso o canal já esteja encerrado, a não ser que uma exceção seja passada por parâmetro nos métodos Complete e TryComplete, vistos acima. No exemplo do código abaixo, o consumidor (reader) permanece ativo enquanto o loop for verdadeiro, e isso acontecerá indefinidamente a não ser que o canal seja encerrado. Enquanto o canal não for encerrado, mas também não chegarem mensagens, ele ficará inativo. A implementação conjunta com o TryRead é interessante pelo seguinte fato: num cenário onde múltiplos consumidores competem por uma mesma mensagem, e a produção dessas mensagens é feita de forma aleatória, a mensagem que foi sinalizada como disponível para todos os consumidores pode ser consumida mais rapidamente por um outro consumidor concorrente.

while (await channel.Reader.WaitToReadAsync())
{
    if (channel.Reader.TryRead(out string msg))
    {                    
        Console.WriteLine(msg);
    }
    else
    {
        Console.WriteLine($"Message already {msg} consumed");
    }
}

3) WriteAsync/ReadAsync: são a implementação assíncrona dos métodos TryRead/TryWrite, com um importante adendo de que ambos disparam uma exceção do tipo ChannelClosedException se o canal já tiver sido encerrado por um writer. Para se evitar essa situação, é preferível usar os métodos WaitTo.

private static async Task ReadAsync(Channel<int> channel, int delay)
{
    while (true)
    {
        var message = await channel.Reader.ReadAsync();
        Console.WriteLine($"Readed {message}");
        await Task.Delay(delay).ConfigureAwait(false);
    }
    // An exception will be thrown when the channel is closed
}

Consumindo todas as mensagens de uma vez

Além dos pares de métodos acima, temos ainda o ReadAllAsync que cria um IAsyncEnumerable de todos os dados disponíveis para leitura no canal e nos permite iterar de maneira assíncrona sobre eles.

Voltando ao AllowSynchronousContinuations

Como dito anteriormente, a propriedade AllowSynchronousContinuations permite com que um produtor vire temporariamente um consumidor. Vamos pensar no seguinte exemplo: um consumidor chama uma operação de leitura num canal sem mensagens, internamente esse canal cria algo como um callback que será chamado quando alguma mensagem for escrita nele e, por default, isso é feito de maneira assíncrona enfileirando a invocação desse callback para ser executada em uma thread diferente da thread do produtor.

Quando setamos essa propriedade como true, estamos dizendo ao canal que esse callback pode ser feito de maneira síncrona, ou seja, a mesma thread que escreveu a mensagem irá consumí-la. Isso pode ser uma vantagem em termos de performance, mas deve ser usada com cuidado em alguns cenários. Por exemplo, se algum tipo de lock estiver sendo usado na escrita, a chamada do callback de leitura pode ocorrer com esse lock ativo, o que pode gerar comportamentos não-desejados no seu código.

Exemplos

Os exemplos abaixo podem ser encontrados neste repo. Nele temos implementações básicas de um produtor e um consumidor interagindo através de um channel de maneira assíncrona e não-bloqueante.

Escrevendo num canal com a capacidade máxima

No exemplo abaixo, setamos velocidades diferentes de leitura e escrita para que a escrita tente acontecer em momentos em que o canal estiver em sua capacidade máxima.

private static async Task ChannelOutOfCapacityExample()
{
    // Setting a read delay bigger than a wrote delay, so that we can see what happens when channels are "full"
    const int readDelay = 500;

    const int writeDelay = 100;

    // Creating a bounded channel with capacity of 1 
    var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(1)
    {
        // Setting this property we say that when the channel is full and another item is dispatched to it, it should wait until the current item is processed to process the next
        FullMode = BoundedChannelFullMode.Wait
    });

    // Calling Task.Run so that the Channel.Writer executes in a different synchronization context than the Channel.Reader
    _ = Task.Run(async () =>
    {
        for (int i = 0; ; i++)
        {
            if (!channel.Writer.TryWrite(i))
            {
                ExtendedConsole.WriteLine($"Dropping {i}", ConsoleColor.Red);
            }

            await Task.Delay(writeDelay).ConfigureAwait(false);
        }
    });

    while (true)
    {
        var message = await channel.Reader.ReadAsync().ConfigureAwait(false);

        ExtendedConsole.WriteLine($"Readed {message}", ConsoleColor.Green);

        await Task.Delay(readDelay).ConfigureAwait(false);
    }
}

Encerrando um canal sem exceções pelo consumidor

No exemplo abaixo exemplificamos o uso do WaitToReadAsync para mostrar como exceções podem ser evitadas no completion do canal.

private static async Task ChannelCompletedWithoutExceptionRaised()
{
    var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(1)
    {
        FullMode = BoundedChannelFullMode.Wait
    });

    _ = Task.Run(async () =>
    {
        for (int i = 0; ; i++)
        {
            await channel.Writer.WriteAsync(i);
            ExtendedConsole.WriteLine($"Writing {i}", ConsoleColor.Blue);

            if (i == 10)
            {
                ExtendedConsole.WriteLine($"Writer: completing channel after 10 executions", ConsoleColor.Yellow);
                channel.Writer.TryComplete();
            }
        }
    });

    // Using WaitToRead, no exception is raised when channel is completed, unless it is explicit passed on completion
    while (await channel.Reader.WaitToReadAsync(default).ConfigureAwait(false))
    {
        if (channel.Reader.TryRead(out int msg))
        {
            ExtendedConsole.WriteLine($"Readed {msg}", ConsoleColor.Green);
        }
        else
        {
            Console.WriteLine($"Message already {msg} consumed");
        }
    }
}

Consumidores concorrentes

No exemplo abaixo, temos um canal com um produtor e três consumidores concorrentes. Dessa forma, quando um consumidor é ativado pelo WaitToReadAsync, por existirem outros consumidores para o mesmo canal, a mensagem que ativou o método já pode ter sido consumida por outro concorrente, o que faria o TryRead retornar false.

private static async Task ChannelWithCompetingConsumers()
{
    var channel = Channel.CreateBounded<int>(new BoundedChannelOptions(1)
    {
        FullMode = BoundedChannelFullMode.Wait
    });

    _ = Task.Run(async () =>
    {
        for (int i = 0; ; i++)
        {
            await channel.Writer.WriteAsync(i);
            ExtendedConsole.WriteLine($"Writing {i}", ConsoleColor.Blue);

            if (i == 50)
            {
                ExtendedConsole.WriteLine($"Writer: completing channel after 10 executions", ConsoleColor.Yellow);
                channel.Writer.TryComplete();
            }
        }
    });

    var firstConsumer = new Consumer<int>(channel.Reader);
    var secondConsumer = new Consumer<int>(channel.Reader);
    var thirdConsumer = new Consumer<int>(channel.Reader);

    await Task.WhenAll(
        firstConsumer.ConsumeAsync(default),
        secondConsumer.ConsumeAsync(default),
        thirdConsumer.ConsumeAsync(default)).ConfigureAwait(false);
}


private class Consumer<T>
{
    private readonly ChannelReader<T> _reader;
    public Guid Id { get; }
    private readonly Random _random = new Random(100);

    public Consumer(ChannelReader<T> reader)
    {
        Id = Guid.NewGuid();
        _reader = reader;
    }

    public async Task ConsumeAsync(CancellationToken cancellationToken)
    {
        // Using WaitToRead, no exception is raised when channel is completed, unless it is explicit passed on completion
        while (await _reader.WaitToReadAsync(cancellationToken))
        {
            if (_reader.TryRead(out T msg))
                ExtendedConsole.WriteLine($"ID {Id} === Readed {msg}", ConsoleColor.Green);
            else
                ExtendedConsole.WriteLine($"ID {Id} === Consumer awoken but message already consumed", ConsoleColor.Yellow);

            await Task.Delay(_random.Next(500));
        }
    }
}

Conclusão

Como visto, channels são abstrações bem poderosas pra mecanismos de pub/sub de maneira assíncrona e não-bloqueante com aplicações .NET (Core e Framework). Sua implementação tem como foco trabalhar em cenários concorrentes com excelentes performance e flexibilidade e são um recurso bem interessante, principalmente pra processamentos feitos de maneira assíncrona através da comunicação entre várias tasks.

Referências

Working with Channels in .NET: https://www.youtube.com/watch?v=gT06qvQLtJ0
An Introduction to System.Threading.Channels: https://devblogs.microsoft.com/dotnet/an-introduction-to-system-threading-channels/
An Introduction to System.Threading.Channels (mesmo título, outro post): https://www.stevejgordon.co.uk/an-introduction-to-system-threading-channels
Exploring System.Threading.Channels: https://ndportmann.com/system-threading-channels/

(pt-BR) Introdução ao RabbitMQ com .NET Core

Mateus Viegas — Mon, 16 Dec 2019 11:49:07 +0000

tl/dr: preparei aqui um guia introdutório a partir de várias leituras e vídeos que escutei ao longo desse ano (fontes serão devidamente dadas) explicando o básico sobre o RabbitMQ e como ele pode ser um recurso interessante na infra-estrutura pra desacoplamento e comunicação entre aplicações. Meu objetivo aqui é ser mais coeso e objetivo e apresentar uma visão básica, porém suficiente pra um entendimento incial, e em português. Caso você saiba inglês, colocarei também alguns links ao longo do artigo.

RabbitMQ: conceitos básicos e motivação

Atualmente, muito se fala do RabbitMQ e message brokers como ferramentas para quebrar os "monolitos" - e aqui eu falo sem o pejorativismo que essa palavra carrega ultimamente. A distribuição de um monolito e o surgimento de arquiteturas distribuídas podem trazer vantagens como desacoplamento, maior coesão e autonomia das partes desacopladas. Essa nova arquitetura distribuída passa a se basear na comunicação entre os participantes para realizar certas ações, o que traz algumas outras complexidades. É introduzida uma camada extra, a rede, e por isso garantias de resiliência, confiabilidade e escalabilidade passam a ser ainda mais necessárias pra esse novo modo de comunicação.

Dito isso, o que é o RabbitMQ? É um message broker, um recurso que se baseia no conceito de mensageria e que funciona da seguinte forma: uma determinada aplicação publica uma determinada mensagem para o broker que recebe essa mensagem e a encaminha para uma fila. Na outra ponta, uma outra aplicação se inscreve para receber as mensagens daquela fila. Temos então alguém que publica (publisher), alguém que se inscreve pra ouvir as mensagens publicadas (subscriber) e um agente no meio que distribui e entrega essas mensagens (message broker). É o padrão pub/sub.

O RabbitMQ traz suporte à alguns protocolos diferentes, como AMQP, MQTT e STOMP. Meu objetivo aqui é ser mais introdutório, mas o Luiz Carlos Faria fala um pouco mais à fundo sobre isso nesse vídeo e numa série de posts que começa com esse aqui.

Vantag ... ou melhor, algumas considerações

Escalabilidade: esse tipo de infra-estrutura suporta centenas de milhares produtos/consumidores de forma simultânea e possui a entrega única como característica, ou seja, cada mensagem só entregue à um consumidor.
Mediação: uma mesma mensagem talvez precise ser entrege para n consumidores que irão processá-la de maneira diferente. O Rabbit possui mecanismos para roteamento e distribuição dessa mensagem, atuando como um mediador dentro da camada de transporte dessa arquitetura distribuída.
Assincronicidade: o envio de mensagens assíncronas retira um ponto de falha de quem envia essas mensagens, o que deixa a aplicação que publica uma mensagem mais leve, pois desacoplou uma responsabilidade dela. Isso contudo, introduz outros possíveis pontos de falha que devem ser tratados para garantir uma aplicação resiliente. Por exemplo, ao retirar o salvamento de log de uma transação SQL dentro da sua aplicação e passando a publicar esse log num message broker pra ser consumido por outra ponta, você retira a dependência dessa transação mas inclui novos possíveis pontos de falha como, por exemplo, a rede que passa a ser uma camada de comunicação e o cliente que vai consumir as mensagens. Em um desses pontos conseguimos ter razoável controle quando publicamos, no outro não. Existem técnicas pra lidar com isso, mas não é o objetivo desse post.

Como funciona

Uma aplicação publica uma mensagem para o Rabbit. Essa mensagem então é enviada para uma exchange, que é um artefato de roteamento, fazendo uma analogia, como um carteiro responsável por fazer a entrega). As exchanges, cada qual de acordo com sua configuração, direcionam a mensagem para uma ou mais filas. Na outra ponta, existe um (ou mais) consumidor(es) responsável por escutar a fila e consumir a mensagem (mais sobre isso abaixo).

É importante dizer que se uma mensagem chegar numa fila com 3 consumidores conectados, apenas um deles receberá a mensagem para processamento (entrega única). Caso seja necessário que mais de um consumidor receba a mesma mensagem, , a exchange para qual a mensagem é enviada deve ser configurada para rotear essa mensagem para várias filas. Assim, a mensagem é replicada e entregue para n filas, permitindo consumidores independentes de uma mesma mensagem - spoiler do que vamos falar mais à frente. Aqui embaixo uma imagem do CloudAMQP para ilustrar:

O CloudAMQP (um provider de Rabbit na nuvem) tem uma série muito boa (em inglês) de RabbitMQ para iniciantes. Disponível nesse link aqui.

Quem faz o que dentro do RabbitMQ

Exchange

Sendo um pouco repetitivo, a Exchange é um artefato de roteamento que funciona como um carteiro responsável por entregar as mensagens. Quando uma mensagem é publicada numa exchange, é enviada uma propriedade (setada por quem envia) chamada routing key. Essa key funciona como o endereço do destinatário. A exchange olha a key e sabe para onde entregar.

Dito isso, Existem quatro tipos exchange no RabbitMQ:

Direct: a mensagem é enviada para uma fila com o mesmo nome da routing key. Se a routing key não for informada, ela é enviada para uma fila padrão.
Fanout: a mensagem é distribuída para todas as filas associadas (falaremos sobre isso mais abaixo). Esse tipo de exchange ignora a routing key.
Topic: a mensagem é distribuída de acordo com um padrão enviado pela routing key.
Header: a mensagem é distribuída de acordo com seus cabeçalhos. Dessa forma, é feito um match com qual consumidor deve receber aquela mensagem.

Queues (filas)

Recebem as mensagens da exchange e armazenam até que um consumidor se conecte para retirar as mensagens de lá. O nome sugere, isso é feito seguindo uma lógica FIFO (first-in, first-out). Podem ter os seguintes atributos:

Durable: a fila sobrevive ao restart do message broker. O RabbitMQ possui um banco de dados chamado Mnesia aonde armazena essas informações.
Exclusive: possui somente 1 consumidor e morre quando não há mais consumidores.
Autodelete: morre quando não há mais mensagens.

Binds (associações)

Para que uma exchange entregue uma mensagem para uma fila, deve haver uma associação, um bind entre elas. Isso pode ser feito de maneira programática por quem envia ou através de uma interface web que o RabbitMQ disponibiliza para gerenciamento do broker.

Virtual hosts (vhosts)

Tudo isso que falamos acima, são entidades do RabbitMQ. Um grupo lógico dessas entidades é chamado de virtual host. Um vhost funciona como um tenant dentro do RabbitMQ, que, inclusive, é descrito (na doc sobre vhosts)[https://www.rabbitmq.com/vhosts.html] por essa razão um sistema multi-tenant.

É importante frisar que um vhost é um separação lógica e não física dessas entidades. Separar esses recursos fisicamente é detalhe de implementação de infra-estrutura.

Fire and forget 🤘🏼

Uma das premissas básicas desse tipo de recurso é que quem publica não conhecem quem consome. Quando uma mensagem é disparada, ela é entregue ao mediador e quem disparou não sabe mais sobre o seu paradeiro. Conseguimos saber apenas se ela chegou ao mediador. Por isso, sempre que algum tipo de retorno for necessário, é seguro dizer que o RabbitMQ não é o recurso adequado para ser utilizado.

Por outro lado, existem alguns lugares aonde podemos ter ganhos consideráveis com esse padrão:

Quando não me importa o retorno da mensagem.
Quando n workers precisam processar uma única mensagem, por exemplo, uma api REST que recebe uma requisição e distribui ela para processamento paralelo.

Quando um consumidor recebe uma mensagem, ele é responsável por dizer ao broker que essa mensagem foi consumida e que já pode ser retirada da fila. É o conceito de acknowledgement. O consumidor é responsável por emitir um ack (mais simpático, mais curto). Ao receber o ack, o Rabbit entende que o consumidor disse: ok Rabbit, já peguei e processei a mensagem, pode retirar ela da fila.

Padrões de consumo

Existem diversos padrões de consumo disponíveis, mas não é meu objetivo falar mais sobre eles nesse post. O RPC é um deles e o próprio site de docs do RabbitMQ tem um tutorial disponível aqui. Eu, contudo, recomendo que os outros tutoriais sejam feitos antes pra exercitar.

E agora, como eu testo isso tudo?!

Provisionando

De maneira rápida, existem duas maneiras que eu recomendo pra se começar a utilizar o RabbitMQ para aprender desenvolvendo.

Conta free no CloudAQMP. Importante dizer que esse post não é um ads pra eles, embora eu já os tenha citado algumas vezes. É só porque eles tem um vasto conteúdo pra iniciante e permitem uma conta free. A desvantagem da conta free é que ela funciona dentro de uma infra-estrutura compartilhada do RabbitMQ, como um virtual host (um assunto-futuro, quem sabe).
Subindo um container no Docker. Eu costumo ter isso nos meus docker-compose.override.yml, mas um comando para subir de primeira é esse aqui:

docker run -d --hostname my-rabbit --name some-rabbit -p 5672:5672 15672:15672 rabbitmq:3-management

O usuário e senha padrão para conexão e acesso da GUI são guest/guest.

Utilizando o RabbitMQ com .NET Core

Existem uma série de bibliotecas que implementam com boas práticas um service bus e trabalham algumas garantias como resiliência para falhas na conexão com o broker, mas não é o objetivo mostrá-las. Pra esse exemplo, basta criar duas aplicações console em .NET Core a adicionar a cada uma o package RabbitMQ.Client pelo seguinte comando:

dotnet add package RabbitMQ.Client

Publisher: console app em .NET Core

Importante: nos dois códigos abaixo vocês vão ver que eu não especifiquei o vhost na conexão. Quando isso acontece, é criada uma conexão ao vhost padrão /.

using System;
using System.Text;
using RabbitMQ.Client;

namespace RabbitMQDemo.Publisher
{
    class Program
    {
        static void Main(string[] args)
        {
            var connectionFactory = new ConnectionFactory()
            {
                HostName = "localhost",
                Port = 5672,
                UserName = "guest",
                Password = "guest",
            };

            using (var connection = connectionFactory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                while (true)
                {
                    Console.WriteLine("Type your message");

                    var teste = Console.ReadLine();

                    channel.QueueDeclare(
                        queue: "tests",
                        durable: false,
                        exclusive: false,
                        autoDelete: false,
                        arguments: null);

                    string message =
                        $"{DateTime.Now.ToString("dd/MM/yyyy HH:mm:ss")} - " +
                        $"Message content: {teste}";
                    var body = Encoding.UTF8.GetBytes(message);

                    channel.BasicPublish(exchange: "",
                                         routingKey: "tests",
                                         basicProperties: null,
                                         body: body);
                }
            }
        }
    }
}

Consumer: console app .NET Core

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace RabbitMQDemo.Consumer
{
    public class Program
    {
        static void Main(string[] args)
        {
            var connectionFactory = new ConnectionFactory()
            {
                HostName = "localhost",
                Port = 5672,
                UserName = "guest",
                Password = "guest"
            };

            using (var connection = connectionFactory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(
                    queue: "tests",
                    durable: false,
                    exclusive: false,
                    autoDelete: false,
                    arguments: null);

                var consumer = new EventingBasicConsumer(channel);

                consumer.Received += (sender, eventArgs) =>
                {
                    var message = Encoding.UTF8.GetString(eventArgs.Body);

                    Console.WriteLine(Environment.NewLine + "[New message received] " + message);
                };

                channel.BasicConsume(queue: "tests",
                     autoAck: true,
                     consumer: consumer);

                Console.WriteLine("Waiting messages to proccess");
                Console.WriteLine("Press some key to exit...");
                Console.ReadKey();
            }

        }
    }
}

Output das aplicações de exemplo

Resumo

O objetivo desse post foi dar um overview básico e o caminho das pedras pra quem quiser se aventurar nesse mundo de sistemas distribuídos com brokers de mensageria. Existem uma série de outras preocupações que esse tipo de arquitetura trazem como: o message broker passa a ser uma parte vital de sua infraestrutura, então ele precisa ter disponibilidade na última casa decimal dos 99,99999....%.

Por fim, gostaria de lembrar que não existem balas de prata. Por isso, pense com cuidado e avalie se cabe na sua aplicação. O RabbitMQ é um recurso incrível, assim como são arquiteturas distribuídas, mas que pode(m) trazer sérias dores de cabeça e prejuízos para o negócio se aplicados sem necessidade e/ou de forma errada.

Forem: Mateus Viegas

Messaging: an Introduction

tl/dr

Loose coupling, but why?

Temporal coupling

Message Types

Commands

Events

Channels

First trade-offs

Wrap-up

A brief brainstorm about the open-closed principle and event-driven architectures

tl/dr

Giving meaning to what it seems obvious

Extensibility with an Event-Driven Architecture

Going further with Domain-Driven Design

Outcome

Explorando System.Threading.Channels

tl/dr

Produtores x Consumidores

Dissecando um channel

Abrindo channels

Encerrando channels

Produzindo e consumindo mensagens

Consumindo todas as mensagens de uma vez

Voltando ao AllowSynchronousContinuations

Exemplos

Escrevendo num canal com a capacidade máxima

Encerrando um canal sem exceções pelo consumidor

Consumidores concorrentes

Conclusão

Referências

(pt-BR) Introdução ao RabbitMQ com .NET Core

RabbitMQ: conceitos básicos e motivação

Vantag ... ou melhor, algumas considerações

Como funciona

Quem faz o que dentro do RabbitMQ

Exchange

Queues (filas)

Binds (associações)

Virtual hosts (vhosts)

Fire and forget 🤘🏼

Padrões de consumo

E agora, como eu testo isso tudo?!

Provisionando

Utilizando o RabbitMQ com .NET Core

Publisher: console app em .NET Core

Consumer: console app .NET Core

Output das aplicações de exemplo

Resumo

Links úteis: