Як створити подієво-орієнтовані додатки у .NET з Kafka та KafkaFlow

Сучасні розподілені системи обробляють мільйони подій щосекунди. Ми бачимо це у соціальних мережах, платіжних системах та інших високонавантажених додатках.

Створення надійних подієво-орієнтованих додатків у .NET вимагає правильного набору інструментів та підходів. Apache Kafka разом з KafkaFlow надає потужне рішення для побудови масштабованих систем в сучасній екосистемі .NET. З випуском .NET 8 та останніх версій .NET Core, ці інструменти стали ще потужнішими та зручнішими у використанні.

У цій статті ми розглянемо покрокове створення подієво-орієнтованого додатка, починаючи від налаштування середовища розробки до розгортання у production. Ви дізнаєтесь як ефективно працювати з повідомленнями, масштабувати систему та забезпечувати її надійність.

Налаштування середовища розробки

Для початку роботи з подієво-орієнтованими додатками нам потрібно правильно налаштувати середовище розробки. Розглянемо покроково всі необхідні компоненти та їх конфігурацію.

Встановлення необхідних компонентів

Для успішної розробки нам знадобляться такі компоненти:

1. .NET SDK (версія 6.0 або вище)
2. Docker Desktop для локального розгортання Kafka
3. Java Runtime Environment (JRE)
4. Visual Studio 2022 або інше IDE за вибором

Базова конфігурація Kafka

Ми будемо використовувати Docker для розгортання Kafka.

Створимо файл docker-compose.yml з базовою конфігурацією:

version: '3'

services:

  zookeeper:

    image: confluentinc/cp-zookeeper

    ports:

      - "2181:2181"

    environment:

      ZOOKEEPER_CLIENT_PORT: 2181

     

  kafka:

    image: confluentinc/confluent-local

    ports:

      - "9092:9092"

    environment:

      KAFKA_BROKER_ID: 1

      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181

      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

Інтеграція KafkaFlow у .NET-проєкт

Для інтеграції KafkaFlow у наш .NET-проєкт ми додамо необхідні NuGet-пакети:

dotnet add package KafkaFlow

dotnet add package KafkaFlow.Microsoft.DependencyInjection

dotnet add package KafkaFlow.LogHandler.Console

dotnet add package KafkaFlow.Serializer.JsonCore

У файлі Program.cs налаштуємо базову конфігурацію KafkaFlow:

services.AddKafka(kafka => kafka

    .UseConsoleLog()

    .AddCluster(cluster => cluster

        .WithBrokers(new[] { "localhost:9092" })

        .CreateTopicIfNotExists("test-topic", 1, 1)

    )

);

Важливо зазначити, що KafkaFlow надає зручний рівень абстракції над клієнтом Confluent .NET Kafka. Це дозволяє нам легко керувати продюсерами та консьюмерами, налаштовувати серіалізацію та створювати масштабовані рішення.

Після завершення налаштування середовища ми можемо перевірити його працездатність, запустивши Docker-контейнери та переконавшись, що наш .NET додаток успішно підключається до Kafka.

Робота з повідомленнями

У світі .NET-розробки ефективна робота з повідомленнями є ключовим елементом побудови надійних подієво-орієнтованих систем. Розглянемо, як KafkaFlow допомагає нам спростити цей процес.

Створення продюсерів повідомлень

Для виправляння повідомлень у Kafka ми використовуємо продюсери. KafkaFlow надає зручний інтерфейс для їх налаштування:

services.AddKafka(kafka => kafka

    .AddCluster(cluster => cluster

        .AddProducer("product-events", producer => producer

            .DefaultTopic("my-topic")

            .AddMiddlewares(m => m

                .AddSerializer<JsonCoreSerializer>()

            )

        )

    )

);

Після налаштування, виправляння повідомлень стає простою операцією:

await producer.ProduceAsync(

    "my-topic",

    Guid.NewGuid().ToString(),

    new ProductEvent { Id = 1, Name = "Test" }

);

Реалізація обробників повідомлень

В .NET 8 та сучасних версіях .NET Framework обробка повідомлень здійснюється через типізовані обробники. Ми створюємо клас-обробник:

public class ProductEventHandler : IMessageHandler<ProductEvent>

{

    public Task Handle(IMessageContext context, ProductEvent message)

    {

        Console.WriteLine($"Отримано подію продукту: {message.Id}");

        return Task.CompletedTask;

    }

}

Серіалізація та десеріалізація даних

KafkaFlow підтримує різні формати серіалізації, що робить його гнучким для різних сценаріїв використання.

Основні опції включають:

1.     JSON серіалізація:

a.     найбільш поширений формат;

b.     простий для налагодження;

c.     підтримка Schema Registry.

2.     Protobuf та Avro:

a.     компактний формат даних;

b.     висока продуктивність;

c.     строга типізація схем.

Налаштування серіалізації виконується через middleware:

.AddMiddlewares(m => m

    .AddSerializer<JsonCoreSerializer>()

    .AddTypeResolver<DefaultTypeResolver>()

)

Важливо зазначити, що KafkaFlow автоматично обробляє tombstone записи та підтримує еволюцію схем через Schema Registry. Це особливо корисно при розробці масштабованих систем на платформі .NET Core.

Масштабування та продуктивність

Оптимізація продуктивності є критичним аспектом розробки подієво-орієнтованих систем у .NET-середовищі. Ми розглянемо як максимально ефективно використовувати можливості KafkaFlow для досягнення високої пропускної здатності.

Паралельна обробка повідомлень

У KafkaFlow ми маємо потужні інструменти для паралельної обробки повідомлень. Основні параметри для оптимізації включають:

services.AddKafka(kafka => kafka

    .AddCluster(cluster => cluster

        .AddConsumer(consumer => consumer

            .Topic("high-load-topic")

            .WithGroupId("processing-group")

            .WithBufferSize(100)

            .WithWorkersCount(10)

        )

    )

);

При налаштуванні паралельної обробки важливо враховувати batch.size (рекомендовано 100000-200000) та linger.ms (10-100 мс) для оптимального балансу між латентністю та пропускною здатністю.

Налаштування пулу воркерів

Ми використовуємо два основні підходи до розподілу навантаження:

1.     BytesSum стратегія:

a.     зберігає порядок повідомлень;

b.     оптимальна для послідовної обробки;

c.     невеликий вплив на продуктивність.

2.     FreeWorker стратегія:

a.     висока швидкість обробки;

b.     відсутність гарантії порядку;

c.     ідеальна для незалежних повідомлень.

.WithWorkerDistributionStrategy<BytesSumDistributionStrategy>()

.WithWorkersCount(Environment.ProcessorCount * 2)

Моніторинг продуктивності

Для ефективного моніторингу продуктивності KafkaFlow надає набір метрик:

Основні метрики:

• messaging.kafka.network.tx
• messaging.kafka.network.rx
• messaging.publish.messages
• messaging.receive.messages

Ми використовуємо вбудований адміністративний Web API для моніторингу в реальному часі:

services.AddKafkaFlowHostedService(

    kafka => kafka.EnableTelemetry()

        .AddCluster(cluster => cluster

            .EnableAdminApi(options =>

                options.Port = 8080)

        )

);

У .NET 8 з'явилися нові можливості для моніторингу через OpenTelemetry, що дозволяє нам відстежувати продуктивність системи більш детально. Ми можемо спостерігати за часом обробки повідомлень, затримками та загальною пропускною здатністю системи.

При налаштуванні моніторингу особливу увагу варто приділити метриці partition_flow_apply_duration_seconds, яка відображає час обробки повідомлень у межах одного запиту до Kafka. Якщо ця метрика постійно показує високі значення при високому завантаженні CPU, нам потрібно розглянути можливість збільшення кількості воркерів або оптимізації коду обробки повідомлень.

Обробка помилок та відмовостійкість

Надійність та стабільність є ключовими вимогами для будь-якої подієво-орієнтованої системи. У нашій роботі з KafkaFlow у середовищі .NET ми приділяємо особливу увагу механізмам обробки помилок та забезпеченню відмовостійкості.

Стратегії обробки винятків

В основі нашого підходу до обробки помилок лежить використання middleware. Ми реалізуємо власний обробник помилок:

public class ErrorHandlingMiddleware : IMessageMiddleware

{

    private readonly ILogger _logger;

   

    public async Task Invoke(IMessageContext context, MiddlewareDelegate next)

    {

        try

        {

            await next(context).ConfigureAwait(false);

        }

        catch(Exception ex)

        {

            _logger.Error("Помилка обробки повідомлення", ex);

            context.ConsumerContext.ShouldStoreOffset = false;

        }

    }

}

Механізми повторних спроб

Ми використовуємо багаторівневий підхід до повторних спроб:

1. Runtime Retries:

• миттєві повторні спроби під час обробки batch;
• налаштовуваний інтервал між спробами;
• максимальна кількість спроб на рівні batch.

2. Redrive System:

• повторна обробка через окрему чергу;
• зберігання failed-повідомлень;
• приоритезація обробки.

У .NET 8 ми отримали розширені можливості для налаштування політик повторних спроб. Наприклад, ми можемо використовувати Simple Retry для тимчасових збоїв або Durable Retry для критично важливих повідомлень, які не можна втратити.

Логування та трасування

Для забезпечення прозорості роботи системи ми використовуємо комплексний підхід до моніторингу.

У .NET Framework ми інтегруємо OpenTelemetry для збору метрик та трейсів:

services.AddOpenTelemetry()

    .WithTracing(builder => builder

        .AddKafkaInstrumentation()

        .AddConsoleExporter())

    .WithMetrics(builder => builder

        .AddMeter("KafkaFlow.Metrics")

        .AddConsoleExporter());

Ключові аспекти моніторингу:

• відстеження часу обробки повідомлень;
• моніторинг кількості retry-спроб;
• аналіз причин відмов.

У випадку виникнення помилок наша система автоматично призупиняє споживача та записує детальну інформацію про проблему. Це дозволяє нам швидко реагувати на інциденти та запобігати втраті даних.

При роботі з .NET Core ми також використовуємо вбудовані механізми health checks для моніторингу стану з'єднання з Kafka. Це допомагає нам забезпечити стабільну роботу системи та швидко виявляти проблеми з підключенням.

Важливо зазначити, що наша стратегія обробки помилок тісно пов'язана з бізнес-вимогами. Для деяких сценаріїв ми налаштовуємо максимальну кількість спроб, після яких повідомлення потрапляє до «мертвої черги» для подальшого аналізу. В інших випадках ми забезпечуємо гарантоване доставлення через механізм Forever Retry.

Тестування та розгортання

Тестування є невіддільною частиною розробки надійних подієво-орієнтованих систем.

У нашій практиці з .NET та Kafka ми використовуємо багаторівневий підхід до тестування, який забезпечує впевненість у якості нашого коду.

Модульне тестування обробників

У .NET Framework ми маємо потужні інструменти для модульного тестування Kafka-компонентів. Для тестування продюсерів ми використовуємо MockProducer:

[Fact]

public async Task TestProductEventProducer()

{

    var mockProducer = new MockProducer<string, ProductEvent>(

        new ProducerConfig(),

        new StringSerializer(),

        new JsonSerializer<ProductEvent>()

    );

    var producer = new ProductEventProducer(mockProducer);

    await producer.PublishEvent(new ProductEvent { Id = 1 });

    var producedMessages = mockProducer.History;

    Assert.Single(producedMessages);

    Assert.Equal("product-topic", producedMessages[0].Topic);

}

Для тестування обробників повідомлень ми використовуємо MockProcessorContext:

[Test]

public void TestMessageHandler()

{

    var context = new MockProcessorContext();

    var handler = new ProductEventHandler();

    handler.Init(context);

   

    handler.Process("key", "value");

   

    var forwarded = context.Forwarded();

    Assert.NotEmpty(forwarded);

}

Інтеграційне тестування

В .NET 8 ми отримали розширені можливості для інтеграційного тестування.

Ми використовуємо TestContainers для створення ізольованого тестового середовища:

public class KafkaIntegrationTests : IAsyncLifetime

{

    private readonly KafkaContainer _kafka = new KafkaContainer()

        .WithExposedPorts(9092);

    public async Task InitializeAsync()

    {

        await _kafka.StartAsync();

        Environment.SetEnvironmentVariable(

            "KAFKA_BOOTSTRAP_SERVERS",

            _kafka.GetBootstrapAddress()

        );

    }

}

Ключові аспекти інтеграційного тестування:

• перевірка end-to-end потоку даних;
• тестування відмовостійкості;
• валідація схем повідомлень;
• перевірка масштабування.

Розгортання у production-середовищі

При розгортанні Kafka-додатків у production-середовищі ми дотримуємося наступного чек-листа:

У .NET Core ми використовуємо вбудовані механізми health checks для моніторингу:

services.AddHealthChecks()

    .AddKafkaHealthCheck(new ProducerConfig

    {

        BootstrapServers = "kafka:9092"

    });

Для автоматизації розгортання ми використовуємо CI/CD pipeline, який включає:

steps:

  - name: Build and Test

    run: dotnet test --configuration Release

   

  - name: Deploy to Production

    env:

      KAFKA_BOOTSTRAP_SERVERS: {{ secrets.KAFKA_SERVERS }}

    run: |

      dotnet publish -c Release

      docker-compose up -d

При роботі з Schema Registry в production ми забезпечуємо сумісність схем через автоматизовані перевірки:

public async Task ValidateSchemaCompatibility()

{

    var schemaRegistry = new CachedSchemaRegistryClient(

        new SchemaRegistryConfig

        {

            Url = "http://schema-registry:8081"

        }

    );

   

    var isCompatible = await schemaRegistry

        .TestCompatibilityAsync("topic-name", newSchema);

       

    Assert.True(isCompatible);

}

У .NET 8 ми також отримали нові можливості для моніторингу через вбудовану інтеграцію з Kafka metrics:

services.AddOpenTelemetry()

    .WithMetrics(builder => builder

        .AddMeter("Kafka.Producer")

        .AddMeter("Kafka.Consumer")

        .AddPrometheusExporter());

Висновок

Створення подієво-орієнтованих додатків з використанням .NET та KafkaFlow відкриває широкі можливості для розробки надійних та масштабованих систем. Ми розглянули всі ключові аспекти: від налаштування середовища розробки до розгортання у production.

Наш практичний підхід охопив найважливіші елементи роботи з Kafka у .NET середовищі:

• конфігурацію та налаштування KafkaFlow;
• ефективну обробку повідомлень та серіалізацію даних;
• стратегії масштабування та оптимізації продуктивності;
• механізми обробки помилок та забезпечення відмовостійкості;
• комплексне тестування та безпечне розгортання.

Платформа .NET 8 разом з сучасними інструментами KafkaFlow надає потужний фундамент для створення високонавантажених систем. Використання описаних підходів та практик допоможе вам уникнути типових помилок та створити надійний, масштабований додаток.

Технології постійно розвиваються, тому важливо слідкувати за новими можливостями та оновленнями KafkaFlow та .NET-платформи. Це дозволить вам ефективно використовувати нові функції та постійно вдосконалювати архітектуру ваших додатків.