Какие знаешь стриминги?

Потоковая обработка данных (Streaming) в C#

В контексте C# и бэкенд-разработки под стримингом (streaming) обычно понимают два основных направления: потоковую обработку данных (data streaming) и потоковую передачу контента (content streaming). Я сосредоточусь на первом, так как он наиболее релевантен для backend-разработки, но кратко упомяну и второе.

1. Потоковая обработка данных (Data Streaming)

Это метод обработки данных непрерывным потоком, где данные обрабатываются по мере поступления, а не пакетами. Ключевые технологии и паттерны:

Apache Kafka

Наиболее популярная distributed streaming-платформа. В экосистеме .NET используется клиент Confluent.Kafka (обертка над librdkafka) или KafkaNET.

using Confluent.Kafka;

var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
using var producer = new ProducerBuilder<Null, string>(config).Build();

// Асинхронная отправка сообщения
var message = new Message<Null, string> { Value = "Hello Kafka!" };
var deliveryResult = await producer.ProduceAsync("test-topic", message);

Azure Event Hubs / Service Bus

Управляемые сервисы от Microsoft для обработки событий и сообщений в реальном времени.

using Azure.Messaging.EventHubs.Producer;

await using var producer = new EventHubProducerClient(connectionString, eventHubName);
using var eventBatch = await producer.CreateBatchAsync();

eventBatch.TryAdd(new EventData(Encoding.UTF8.GetBytes("Event data")));
await producer.SendAsync(eventBatch);

RabbitMQ с Streaming Extensions

Хотя RabbitMQ традиционно message broker, с помощью плагинов и паттернов (как Publisher/Consumer) может использоваться для streaming-сценариев.

2. Reactive Extensions (Rx.NET)

Библиотека для композиции асинхронных и событийных программ с помощью observable sequences. Основана на паттерне Observer.

using System.Reactive.Linq;

// Создание observable из событий
var observable = Observable.Interval(TimeSpan.FromSeconds(1))
    .Where(x => x % 2 == 0)
    .Select(x => $"Event {x}");

// Подписка
var subscription = observable.Subscribe(
    Console.WriteLine,
    ex => Console.WriteLine($"Error: {ex}"),
    () => Console.WriteLine("Completed")
);

3. IAsyncEnumerable (C# 8.0+)

Нативный механизм C# для работы с асинхронными потоками данных, идеально подходит для streaming API.

public async IAsyncEnumerable<int> GenerateStreamAsync()
{
    for (int i = 0; i < 10; i++)
    {
        await Task.Delay(100); // Имитация асинхронной операции
        yield return i;
    }
}

// Потребление
await foreach (var item in GenerateStreamAsync())
{
    Console.WriteLine($"Received: {item}");
}

4. gRPC Streaming

Поддержка различных режимов streaming в gRPC:

• Server streaming: сервер отправляет поток сообщений клиенту
• Client streaming: клиент отправляет поток серверу
• Bidirectional streaming: двусторонний поток

service DataService {
  rpc StreamData (stream ClientMessage) returns (stream ServerMessage);
}

// На стороне сервера
public override async Task StreamData(
    IAsyncStreamReader<ClientMessage> requestStream,
    IServerStreamWriter<ServerMessage> responseStream,
    ServerCallContext context)
{
    await foreach (var message in requestStream.ReadAllAsync())
    {
        // Обработка и ответ
        await responseStream.WriteAsync(new ServerMessage());
    }
}

5. Потоковая передача файлов и контента

В ASP.NET Core для эффективной работы с большими файлами:

app.MapPost("/upload", async (HttpRequest request) =>
{
    await using var stream = File.Create("uploaded.bin");
    await request.Body.CopyToAsync(stream); // Потоковая запись
});

app.MapGet("/download", async (HttpContext context) =>
{
    context.Response.ContentType = "application/octet-stream";
    await using var fileStream = File.OpenRead("largefile.bin");
    await fileStream.CopyToAsync(context.Response.Body); // Потоковая отдача
});

6. Apache Spark Streaming (для .NET)

С помощью .NET for Apache Spark можно создавать streaming-приложения для обработки больших данных.

7. Cloud-native streaming

• Azure Stream Analytics: SQL-подобный язык для обработки потоков
• AWS Kinesis: аналог Kafka в экосистеме AWS
• Google Pub/Sub: messaging и streaming от Google

Ключевые преимущества streaming-подходов:

• Низкая задержка (low latency) обработки данных
• Эффективное использование памяти (данные не накапливаются)
• Высокая масштабируемость через партиционирование
• Отказоустойчивость через репликацию и checkpoint-и

Паттерны обработки:

1. Event Sourcing - хранение состояния как потока событий
2. CQRS - разделение на команды (запись) и запросы (чтение)
3. Complex Event Processing (CEP) - анализ паттернов в потоках событий

В современных микросервисных архитектурах streaming-технологии стали стандартом для реализации event-driven архитектур, обеспечивая слабую связанность сервисов и реальное время обработки данных. Выбор конкретной технологии зависит от требований к пропускной способности, задержке, экосистеме и инфраструктуре.