C# Channel<T> — Introdução Producer / Consumer
É muito comum nos depararmos com situações nas quais precisamos implementar mecanismos de “Producer/Consumer”, ou alguma variação deste processo, e quando isso acontecia precisávamos fazê-lo utilizando classes como Queue<T>, Semaphore e outros componentes.
Para nos auxiliar nessas implementações, foi introduzido no .Net Core 3.X o objeto Channel<T>.
Mas antes de detalharmos o funcionamento e utilização dessa classe, vamos a uma breve descrição do que seria “Producer/Consumer ”, para que fique claro o problema que a classe Channel<T> nos ajuda a solucionar.
Producer / Consumer
Em computação, chamamos de “Producer / Consumer” o processo onde temos um ator, producer, responsável por gerar e inserir itens (mensagens) em uma fila, e outro ator/atores, chamado de consumer, responsável por remover esses itens da fila e executar algum trabalho com eles.
Vamos utilizar a imagem abaixo para ilustrar melhor essa operação.
Vamos dividir esse processo em três partes, como enumerado na imagem.
- Produtor: processo responsável por inserir mensagens na fila. A fonte de dados para a geração das mensagens é irrelevante, ela pode ser um arquivo, banco de dados, uma Api, etc. O importante aqui é a função de inserir mensagens na fila.
- Fila: estrutura de dado responsável por receber mensagens, armazená-las, e entregá-las quando solicitado. A mensagem é entregue somente uma vez, evitando assim processamento em duplicidade. Na imagem foi representado uma fila FIFO, First in — First out, mas existem outros tipos.
- Consumidor: processo responsável por remover uma mensagem da fila e realizar seu processamento. Podem existir um ou mais consumidores retirando mensagens da mesma fila, isso lhe proporciona escalabilidade.
Agora que estamos alinhados com a definição e funcionamento do “Producer / Consumer”, vamos abordar como o Channel<T> pode nos auxiliar nessa implementação.
Channel<T>
É uma estrutura de dado desenvolvida para gerenciar a passagem de dados entre produtores e consumidores. Ela esta disponível no namespace System.Threading.Channels e possui algumas vantagens em relação a outros componentes semelhantes para controle de filas em .Net, como por exemplo Queue<T>. Entre as vantagens podemos citar:
- Abstrai a complexidade de gerenciamento da comunicação entre producer e consumer, por exemplo, notificação de fim das publicações, limite de itens na fila, e outras configurações;
- melhor especialização dos componentes, pois existem componentes específicos para escrita, producer, e componentes específicos para leitura, consumer;
- Tradesafe, ou seja, múltiplas threads podem realizar a escrita e leitura no mesmo objeto.
Mas vamos ao código para demonstrar a utilização desse componente.
Show me the code!
Em todos os exemplos vamos utilizar a classe abaixo para representar nossa mensagem.
Simples Producer / Consumer
Podemos observar no código acima que o objeto channel encapsula a lógica de notificação de fim dos itens, lançando a exceção ChannelClosedException quando não existem mais itens disponíveis na fila. Essa notificação ocorre pois o producer realizou a notificação de fim de publicação de itens na fila, isso e realizado na linha 14, com a instrução Writer.Complete().
E se não quisermos a notificação de fim das mensagens no channel por meio de uma exception, como podemos fazer? Para isso a classe ChannelReader possui o método ReadAllAsync que retorna uma IAsyncEnumerable, que pode ser iterada por um “await for”.
Especializando melhor os componentes
Num cenário onde construímos uma implementação de producer/consumer utilizando o objeto Queue<T>, tanto o objeto responsável pela produção das mensagens, quanto o objeto responsável por consumir as mensagens devem ter acesso ao objeto Queue<T>, e nada impediria que o objeto consumer também fizesse a inclusão de novas mensagens na fila, já que ele tem acesso ao objeto Queue<T>, ou seja, não temos uma estrutura responsável somente pela leitura das mensagens.
Já o objeto Channel<T> possibilita essa segmentação entre produtores e consumidores de maneira facilita, já que a classe Channel possui a propriedade Reader, do tipo ChannelReader, responsável pela leitura, e a propriedade Writer, do tipo ChannelWriter, responsável pela escrita. Veja como ficaria um exemplo simples dessa segmentação.
Esses códigos estão disponíveis em meu GitHub.
Considerações Finais
Já tive que implementar um mecanismo de producer/consumer utilizando Queue/ConcurrentQueue, Semaphore e CancellationToken, e sei o trabalho que isso deu, e quando me deparei com a implementação do Channel vi que isso poderia facilitar bastante nossa vida ;)
Existem outras configurações no channel que podem adequá-lo ainda mais às suas necessidades, como o Channel Bounded, que possibilita configurarmos um limite de itens na fila, mas meu objetivo era trazer essa introdução de utilização.
Espero ter ajudado e até a próxima!
