10 дней .Net Aspire: День 5 — Интеграция Apache Kafka

Пошаговое руководство по использованию компонента .Net Aspire Kafka в Visual Studio.

.Net Aspire Framework используется для разработки облачных и готовых к производству распределенных приложений. Он состоит из компонентов для обработки облачных задач, таких как Redis, Postgres и т. д.

Предпосылки

• Установить.Net 8
• УстановитьVisual Studio 2022версия 17 или выше
• Рабочая нагрузка .Net Aspire
• Среда выполнения контейнера, напримерРабочий стол Docker

Цели

Узнайте, как создать начальный проект с использованием .Net Aspire с компонентом Apache Kafka.

Образец Github: Структура решения разделена на следующие проекты:

• DotnetAspireChallenge.ApiService
• DotnetAspireChallenge.AppHost
• DotnetAspireChallenge.ServiceDefaults
• DotnetAspireChallenge.Web

Начиная

Шаг 1: Установите следующий пакет NuGet

Установите следующий пакет Nuget в последующий проект «DotnetAspireChallenge.AppHost”

dotnet add package Aspire.Hosting.Kafka

В проекте выше зарегистрируйте Kafka UI, как показано ниже.

    var messaging = builder.AddKafka("messaging")
```                           .WithKafkaUI();

Then finally add a reference to both the Producer and Consumer where the producer is “DotnetAspireChallenge.ApiService” and the consumer is “DotnetAspireChallenge.Web” project respectively.
```csharp
    var apiService = builder.AddProject<Projects.DotnetAspireChallenge_ApiService>("apiservice")
        .WithReference(messaging);
    
    
    
    builder.AddProject<Projects.DotnetAspireChallenge_Web>("webfrontend")
        .WithExternalHttpEndpoints()
        .WithReference(cache)
        .WithReference(apiService)
        .WithReference(messaging);

Шаг 2: Добавьте зависимость Kafka Producer

Добавьте зависимость в файл Program.cs проекта «**DotnetAspireChallenge.**ApiService»

    builder.AddKafkaProducer<string, string>("messaging");

и добавьте соответствующую минимальную конечную точку API, используя следующий код.

    public static class AspireKafkaExtension
    {
        public static void MapAspireKafkaEndpoint(this WebApplication app)
        {
            app.MapGet("/send", async (IProducer<string, string> services, string key, string value) =>
            {
                try
                {
                    var message = new Message<string, string> { Key = key, Value = value };
                    DeliveryResult<string, string>? result = await services.ProduceAsync("messaging", message);
                    return result;
                }
                catch (Exception ex)
                {
    
                    throw;
                }
    
            });
        }
    }

Конечная точка принимает два параметра, а именно ключ и значение, в качестве значений маршрута и создает сообщение на сервере Kafka, размещенном в Docker.

https://localhost:7313/send?key=key&value=1

Шаг 3: Добавьте зависимость Kafka Consumer

Теперь двигайся «DotnetAspireChallenge.Web» проект, в котором зарегистрироваться как производитель Кафки

    builder.AddKafkaConsumer<string, string>("messaging", options =>
    {
        options.Config.GroupId = "my-consumer-group";
        options.Config.AutoOffsetReset = AutoOffsetReset.Earliest;
        options.Config.EnableAutoCommit = false;
    });

Шаг 4: Создайте страницу Razor

Создайте базовую страницу Razor с именем «KafkaConsumer.razor» для отображения полученного сообщения с сервера Kafka.

    @page "/kafka"
    @attribute [StreamRendering(true)]
    @attribute [OutputCache(Duration = 5)]
    @using Confluent.Kafka
    <h3>KafkaConsumer</h3>
    
    @inject KafkaConsumeMessageClient kafaConsumeMessageClient
    <PageTitle>Kafka Consumed Message</PageTitle>
    
    <h1>Kafka</h1>
    
    <p>This component demonstrates showing data loaded from a backend API service.</p>
    
    @if (consumedMessage == null)
    {
        <p><em>Loading...</em></p>
    }
    else
    {
        <table class="table">
            <thead>
                <tr>
                    <th>Topic</th>
                    <th>Value</th>
                </tr>
            </thead>
            <tbody>
                <tr>
                    <td>@consumedMessage.Topic</td>
                    <td>@consumedMessage.Value</td>
                </tr>
    
            </tbody>
        </table>
    }
    @code {
        private ConsumeResult<string, string>? consumedMessage;
    
        protected override async Task OnInitializedAsync() => consumedMessage = kafaConsumeMessageClient.GetKafkaMessage();
    }

Шаг 5: Настройте HttpCall для ApiService

    public class KafkaConsumeMessageClient(HttpClient httpClient, IConsumer<string, string> _consumer)
    {
    
        public ConsumeResult<string, string>? GetKafkaMessage(CancellationToken cancellationToken = default)
        {
            ConsumeResult<string, string>? deliveryResult = null;
            _consumer.Subscribe("messaging");
            deliveryResult = _consumer.Consume(TimeSpan.FromSeconds(10));
    
            return deliveryResult;
        }
    }

Демоверсия Kafka Produce

Демонстрация пользовательского интерфейса Kafka

Демоверсия Kafka Consume