First touch of Kafka

Прежде чем начать я бы хотел отметить, что это всего лишь небольшой туториал по быстрому старту для тех кто, как и я, ни разу не использовал Kafka на практике

И так приступим!

Единственный брокер Kafka и необходимый для его работы ZooKeeper я буду запускать в Docker

Сперва создам отдельную сеть kafkanet

docker network create kafkanet

Запуск контейнера с ZooKeeper

docker run -d --network=kafkanet --name=zookeeper -e ZOOKEEPER_CLIENT_PORT=2181 -e ZOOKEEPER_TICK_TIME=2000 -p 2181:2181 confluentinc/cp-zookeeper

Запуск контейнера с Kafka

docker run -d --network=kafkanet --name=kafka -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 -p 9092:9092 confluentinc/cp-kafka

Для того чтобы убедиться в отсутствии ошибок, можно вывести лог docker logs kafka

Далее проверю функционирование брокера Kafka, выполнив простые операции, включающие создание тестовой темы, генерацию сообщений и их потребление

Для этого сценария подключусь к контейнеру kafka

docker exec -it kafka bash

Создам топик demo-topic

/bin/kafka-topics --create --topic demo-topic --bootstrap-server kafka:9092

Выведу список всех топиков

/bin/kafka-topics --list --zookeeper zookeeper:2181

И выведу описание созданного топика

/bin/kafka-topics --describe --topic demo-topic --bootstrap-server kafka:9092

Сгенерирую несколько сообщений

/bin/kafka-console-producer --topic demo-topic --bootstrap-server kafka:9092

И после прочитаю эти сообщения

/bin/kafka-console-consumer --topic demo-topic --from-beginning --bootstrap-server kafka:9092

Далее я создам два небольших .NET приложения: KafkaProducer, которое будет генерировать сообщения, и KafkaConsumer, которое будет потреблять сообщения. Для реализации мне понадобятся пакеты Confluent.Kafka и Microsoft.Extensions.Hosting

В проект KafkaProducer добавлю класс KafkaProducerService

using Confluent.Kafka;using Microsoft.Extensions.Hosting;using Microsoft.Extensions.Logging;using System.Threading;using System.Threading.Tasks;namespace KafkaProducer{    public class KafkaProducerService : IHostedService    {        private readonly ILogger<KafkaProducerService> _logger;        private readonly IProducer<Null, string> _producer;        public KafkaProducerService(ILogger<KafkaProducerService> logger)        {            _logger = logger;            var config = new ProducerConfig            {                BootstrapServers = "localhost:9092"            };            _producer = new ProducerBuilder<Null, string>(config).Build();        }        public async Task StartAsync(CancellationToken cancellationToken)        {            for (var i = 0; i < 5; i++)            {                var value = $"Event N {i}";                _logger.LogInformation($"Sending >> {value}");                await _producer.ProduceAsync(                    "demo-topic",                    new Message<Null, string> { Value = value },                    cancellationToken);            }        }        public Task StopAsync(CancellationToken cancellationToken)        {            _producer?.Dispose();            _logger.LogInformation($"{nameof(KafkaProducerService)} stopped");            return Task.CompletedTask;        }    }}

Изменю файл Program.cs

using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.Hosting;using System;namespace KafkaProducer{    class Program    {        static void Main(string[] args)        {            CreateHostBuilder(args).Build().Run();            Console.ReadKey();        }        private static IHostBuilder CreateHostBuilder(string[] args) =>            Host                .CreateDefaultBuilder(args)                .ConfigureServices((context, collection) =>                    collection.AddHostedService<KafkaProducerService>());    }}

В проект KafkaConsumer добавлю класс KafkaConsumerService

using Confluent.Kafka;using Microsoft.Extensions.Hosting;using Microsoft.Extensions.Logging;using System.Threading;using System.Threading.Tasks;namespace KafkaConsumer{    public class KafkaConsumerService : IHostedService    {        private readonly ILogger<KafkaConsumerService> _logger;        private readonly IConsumer<Ignore, string> _consumer;        public KafkaConsumerService(ILogger<KafkaConsumerService> logger)        {            _logger = logger;            var config = new ConsumerConfig            {                BootstrapServers = "localhost:9092",                GroupId = "demo-group",                AutoOffsetReset = AutoOffsetReset.Earliest            };            _consumer = new ConsumerBuilder<Ignore, string>(config).Build();        }        public Task StartAsync(CancellationToken cancellationToken)        {            _consumer.Subscribe("demo-topic");            while (!cancellationToken.IsCancellationRequested)            {                var consumeResult = _consumer.Consume(cancellationToken);                _logger.LogInformation($"Received >> {consumeResult.Message.Value}");            }            return Task.CompletedTask;        }        public Task StopAsync(CancellationToken cancellationToken)        {            _consumer?.Dispose();            _logger.LogInformation($"{nameof(KafkaConsumerService)} stopped");            return Task.CompletedTask;        }    }}

Изменю файл Program.cs

using Microsoft.Extensions.DependencyInjection;using Microsoft.Extensions.Hosting;using System;namespace KafkaConsumer{    class Program    {        static void Main(string[] args)        {            CreateHostBuilder(args).Build().Run();            Console.ReadKey();        }        private static IHostBuilder CreateHostBuilder(string[] args) =>            Host                .CreateDefaultBuilder(args)                .ConfigureServices((context, collection) =>                    collection.AddHostedService<KafkaConsumerService>());    }}

Результат работы приложений (ссылка на репозиторий)