پلتفرم .NET (به‌ویژه نسخه‌های modern شامل .NET 8 و .NET 9) با بهره‌گیری از زیرساخت‌های با کارایی بالا (High-Performance)، اکوسیستم گسترده، پایداری بالا در مقیاس سازمانی (Enterprise Level) و پشتیبانی نیتیو از برنامه‌نویسی همزمان (Asynchronous) و پردازش جریانی، به یکی از قدرتمندترین انتخاب‌ها برای پیاده‌سازی زیرساخت‌های پردازش داده‌های IoT تبدیل شده است.

معماری کلان پردازش داده‌های IoT (IoT Data Pipeline)

یک معماری استاندارد و مقیاس‌پذیر برای داده‌های IoT از چهار لایه اصلی تشکیل شده است:

+-------------------+      +-------------------+      +-------------------+      +-------------------+
|  1. Ingestion     | ---> |   2. Processing   | ---> |   3. Storage      | ---> |  4. Analytics/ML  |
|  (MQTT / Hubs)    |      |  (Reactive/Worker)|      | (Time-Series DB)  |      |   (ML.NET / BI)   |
+-------------------+      +-------------------+      +-------------------+      +-------------------+

1. لایه دریافت داده (Ingestion Layer): دریافت جریان‌های داده‌ای سنگین از حسگرها از طریق پروتکل‌های کم‌حجم مانند MQTT، AMQP یا HTTP REST.
2. لایه پردازش جریانی (Stream Processing Layer): فیلترسازی، اعتبارسنجی، غنی‌سازی (Enrichment) و محاسبه میانگین‌های متحرک یا تشخیص آنومالی روی داده‌های در حال حرکت (Data-in-Motion).
3. لایه ذخیره‌سازی (Storage Layer): ذخیره‌سازی بهینه داده‌های زمانی در پایگاه‌های داده مخصوص (Time-Series Databases) مانند TimescaleDB یا InfluxDB.
4. لایه تحلیل و یادگیری ماشین (Analytics & ML Layer): تحلیل‌های پیشرفته، پیش‌بینی خرابی (Predictive Maintenance) و مصورکننده داده‌ها.

لایه دریافت داده (Data Ingestion) با پروتکل MQTT در .NET

پروتکل MQTT (Message Queuing Telemetry Transport) به دلیل Over-head بسیار پایین هدرهای بسته، ایده‌آل‌ترین گزینه برای ارتباطات حسگرهای IoT است. در پلتفرم .NET، کتابخانه MQTTnet یکی از پرکاربردترین و به‌روزترین ابزارها برای مدیریت ارتباطات MQTT است.

پیاده‌سازی یک Broker/Client بالااستفاده در .NET

برای دریافت داده‌های جریان بالا، ساختار دریافت‌کننده (Subscriber) باید کاملاً غیرهمزمان (Asynchronous) و بدون مسدودسازی نخ‌ها (Non-blocking) طراحی شود.

using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using MQTTnet;
using MQTTnet.Client;

public class MqttTelemetryIngestionService
{
    private IMqttClient _mqttClient;

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        var factory = new MqttFactory();
        _mqttClient = factory.CreateMqttClient();

        var options = new MqttClientOptionsBuilder()
            .WithTcpServer("iot-broker.internal.net", 1883)
            .WithClientId($"IoT_Processor_{Guid.NewGuid()}")
            .WithCleanSession()
            .Build();

        _mqttClient.ApplicationMessageReceivedAsync += HandleIncomingMessageAsync;

        await _mqttClient.ConnectAsync(options, cancellationToken);

        var topicFilter = factory.CreateTopicFilterBuilder()
            .WithTopic("sensors/+/telemetry")
            .WithAtLeastOnceQoS()
            .Build();

        await _mqttClient.SubscribeAsync(topicFilter, cancellationToken);
    }

    private Task HandleIncomingMessageAsync(MqttApplicationMessageReceivedEventArgs e)
    {
        // Extraction without allocation string using Memory/Span where possible
        ReadOnlyMemory<byte> payload = e.ApplicationMessage.PayloadSegment.AsMemory();
        
        // ارسال جهت پردازش غیرهمزمان در خط لوله بعدی (Channel یا Reactive Extensions)
        TelemetryBackgroundQueue.Enqueue(payload);

        return Task.CompletedTask;
    }
}

الگوی High-Throughput با System.Threading.Channels

یکی از مشکلات رایج در دریافت داده‌های حجیم (High Throughput)، ایجاد گلوگاه (Bottleneck) در زمان نوشتن داده‌ها در دیتابیس یا تحلیل آن‌هاست. اگر عملیات تحلیل درون همان نخی انجام شود که پیام را از MQTT دریافت می‌کند، دریافت پیام‌های بعدی به تاخیر افتاده و صف پیام‌ها در سمت Broker سرریز می‌کند.

برای جداسازی (Decoupling) بخش دریافت از بخش پردازش، استفاده از System.Threading.Channels که یک کانال Producer-Consumer بسیار سریع و کم‌هزینه از نظر تخصیص حافظه (Zero-Allocation Producer-Consumer Pattern) است، توصیه می‌شود.

پیاده‌سازی کانال درون‌حافظه‌ای با کانکرنسی بالا

using System;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

public record TelemetryPayload(string DeviceId, double Temperature, double Humidity, DateTime Timestamp);

public class TelemetryChannelPipeline
{
    private readonly Channel<TelemetryPayload> _channel;

    public TelemetryChannelPipeline(int capacity = 100000)
    {
        // استفاده از BoundedChannel برای جلوگیری از OOM (Out of Memory) در حالت Spikes
        var options = new BoundedChannelOptions(capacity)
        {
            SingleWriter = false,
            SingleReader = false,
            FullMode = BoundedChannelFullMode.Wait
        };
        _channel = Channel.CreateBounded<TelemetryPayload>(options);
    }

    public ValueTask PublishAsync(TelemetryPayload item, CancellationToken ct = default)
    {
        return _channel.Writer.WriteAsync(item, ct);
    }

    public async Task StartConsumersAsync(int consumerCount, CancellationToken ct)
    {
        for (int i = 0; i < consumerCount; i++)
        {
            _ = Task.Run(() => ConsumeAsync(_channel.Reader, ct), ct);
        }
    }

    private async Task ConsumeAsync(ChannelReader<TelemetryPayload> reader, CancellationToken ct)
    {
        while (await reader.WaitToReadAsync(ct))
        {
            while (reader.TryRead(out var payload))
            {
                // پردازش داده شامل تحلیل آنومالی، میانگین‌گیری و ذخیره‌سازی
                ProcessTelemetry(payload);
            }
        }
    }

    private void ProcessTelemetry(TelemetryPayload payload)
    {
        // منطق پردازش داده پردازش شده
    }
}

تحلیل پردازش جریانی (Stream Processing) با Reactive Extensions (Rx.NET)

داده‌های IoT دارای ماهیتی زنده و وابسته به زمان (Time-Series Dynamics) هستند. بررسی داده‌ها به صورت مجزا اغلب ارزش عملیاتی کمی دارد؛ ارزش واقعی در پردازش پنجره‌های زمانی (Time Windows) نهفته است (مثلاً: اگر میانگین دمای موتور در ۵ دقیقه گذشته بیش از ۸۵ درجه بود، هشدار صادر کن).

کتابخانه System.Reactive (Rx.NET) ابزار فوق‌العاده‌ای برای پردازش رویداد محور (Event-Driven) و اعمال پنجره‌های زمانی متحرک (Sliding & Tumbling Windows) است.

محاسبه میانگین متحرک و تشخیص آنومالی با Rx.NET

using System;
using System.Reactive.Linq;
using System.Reactive.Subjects;

public class ReactiveTelemetryAnalyzer
{
    private readonly Subject<TelemetryPayload> _stream = new();

    public void OnNextTelemetry(TelemetryPayload payload)
    {
        _stream.OnNext(payload);
    }

    public void InitializePipeline()
    {
        // Tumbling Window: محاسبه میانگین دما در هر ۱۰ ثانیه برای هر دستگاه
        _stream
            .GroupBy(p => p.DeviceId)
            .Subscribe(deviceGroup =>
            {
                deviceGroup
                    .Buffer(TimeSpan.FromSeconds(10))
                    .Where(buffer => buffer.Count > 0)
                    .Subscribe(batch =>
                    {
                        double avgTemp = batch.Average(b => b.Temperature);
                        Console.WriteLine($"[Device {deviceGroup.Key}] 10s Avg Temp: {avgTemp:F2}°C (Readings count: {batch.Count})");

                        if (avgTemp > 90.0)
                        {
                            TriggerAlert(deviceGroup.Key, $"Overheating warning! Average: {avgTemp:F2}");
                        }
                    });
            });

        // Sliding Window: تشخیص نوسان ناگهانی دما (تغییر بیش از ۱۰ درجه در کمتر از ۳ ثانیه)
        _stream
            .GroupBy(p => p.DeviceId)
            .Subscribe(deviceGroup =>
            {
                deviceGroup
                    .Window(TimeSpan.FromSeconds(3))
                    .SelectMany(window => window.ToList())
                    .Subscribe(list =>
                    {
                        if (list.Count >= 2)
                        {
                            var delta = Math.Abs(list[^1].Temperature - list[0].Temperature);
                            if (delta > 10.0)
                            {
                                TriggerAlert(deviceGroup.Key, $"Spike detected: Jumped {delta}°C in 3 seconds!");
                            }
                        }
                    });
            });
    }

    private void TriggerAlert(string deviceId, string message)
    {
        // ارسال به سیستم‌های اعلان فوری مانند SignalR یا SMS Hub
    }
}

بهینه‌سازی کارایی و مدیریت حافظه در مقیاس‌های سنگین (High Performance .NET)

در سامانه‌هایی که روزانه میلیاردها پیام IoT دریافت می‌کنند، ایجاد فشار روی Garbage Collector (GC) به علت تخصیص زیاد حافظه (Allocations)، می‌تواند منجر به توقف‌های کوتاه (GC Pauses) و در نتیجه کندی سیستم شود. برای دستیابی به کارایی بالا، رعایت نکات زیر الزامی است:

الف) استفاده از Span<T>، ReadOnlySpan<T> و Memory<T>

به جای تبدیل بایت‌های دریافتی شبکه (Byte Arrays) به رشته (String) یا شیء، باید کار بر روی آرایه‌های بایت به صورت مستقیم و بدون Allocation با استفاده از Span<T> انجام شود.

ب) استفاده از System.Text.Json به جای Newtonsoft.Json

کتابخانه مدرن System.Text.Json بر مبنای Utf8JsonReader و بدون تخصیص حافظه اضافه (Zero-allocation UTF-8 parsing) طراحی شده است.

using System;
using System.Text.Json;

public static class FastTelemetryParser
{
    // پارس کردن مستقیم از UTF8 بدون تبدیل به String
    public static TelemetryPayload Parse(ReadOnlySpan<byte> utf8Json)
    {
        var reader = new Utf8JsonReader(utf8Json);
        string deviceId = null;
        double temp = 0;
        double humidity = 0;
        DateTime time = default;

        while (reader.Read())
        {
            if (reader.TokenType == JsonTokenType.PropertyName)
            {
                if (reader.ValueTextEquals("deviceId"u8))
                {
                    reader.Read();
                    deviceId = reader.GetString();
                }
                else if (reader.ValueTextEquals("temp"u8))
                {
                    reader.Read();
                    temp = reader.GetDouble();
                }
                else if (reader.ValueTextEquals("humidity"u8))
                {
                    reader.Read();
                    humidity = reader.GetDouble();
                }
                else if (reader.ValueTextEquals("timestamp"u8))
                {
                    reader.Read();
                    time = reader.GetDateTime();
                }
            }
        }

        return new TelemetryPayload(deviceId, temp, humidity, time);
    }
}

ج) استفاده از ArrayPool<T>

برای بافرهای متغیر و ساختارهای داده موقت، به جای ایجاد آرایه‌های جدید با new byte[] از حافظه اشتراکی الگوهای پูล (Pool) با ArrayPool<byte>.Shared استفاده می‌شود تا بار کاری Large Object Heap (LOH) کاهش یابد.

لایه ذخیره‌سازی: درج دسته‌ای (Batch Inserter) در Time-Series Databases

نوشتن تک‌تک داده‌های حسگر در پایگاه داده، Bottleneck بزرگی ایجاد می‌کند. در پردازش داده‌های IoT، روش استاندارد، درج گروهی یا Batch Insertion است.

الگوی پیاده‌سازی Batching Worker در .NET 8

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Hosting;

public class TimescaleDbBatchWriterService : BackgroundService
{
    private readonly ChannelReader<TelemetryPayload> _reader;
    private readonly List<TelemetryPayload> _batchBuffer = new(1000);
    private readonly TimeSpan _flushInterval = TimeSpan.FromSeconds(2);

    public TimescaleDbBatchWriterService(ChannelReader<TelemetryPayload> reader)
    {
        _reader = reader;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        var timer = new PeriodicTimer(_flushInterval);

        while (!stoppingToken.IsCancellationRequested)
        {
            while (_reader.TryRead(out var item))
            {
                _batchBuffer.Add(item);
                if (_batchBuffer.Count >= 1000)
                {
                    await FlushToDatabaseAsync(_batchBuffer, stoppingToken);
                    _batchBuffer.Clear();
                }
            }

            if (await timer.WaitForNextTickAsync(stoppingToken) && _batchBuffer.Count > 0)
            {
                await FlushToDatabaseAsync(_batchBuffer, stoppingToken);
                _batchBuffer.Clear();
            }
        }
    }

    private async Task FlushToDatabaseAsync(List<TelemetryPayload> items, CancellationToken ct)
    {
        // پیاده‌سازی عملیات Npgsql Binary COPY برای PostgreSQL/TimescaleDB با حداکثر سرعت
        // e.g., using (var writer = conn.BeginBinaryImport("COPY telemetry FROM STDIN Binary"))
        await Task.CompletedTask;
    }
}

هوش مصنوعی در Edge و Cloud با ML.NET

یکی از حوزه‌های جدید در IoT، اجرای مدل‌های یادگیری ماشین برای تشخیص آنومالی (Anomaly Detection) و پیش‌بینی نگهداری (Predictive Maintenance) روی داده‌های زنده است. پلتفرم .NET با ارائه فریم‌ورک ML.NET امکان آموزش و اجرای مدل‌های ML را بدون نیاز به وابستگی‌های سنگین پایتون فراهم می‌سازد.

تشخیص آنومالی سری‌های زمانی با ML.NET

using Microsoft.ML;
using Microsoft.ML.TimeSeries;
using System.Collections.Generic;

public class TelemetryAnomalyDetector
{
    private readonly MLContext _mlContext = new MLContext();

    public void DetectSpikes(List<TelemetryDataInput> data)
    {
        IDataView dataView = _mlContext.Data.LoadFromEnumerable(data);

        // استفاده از الگوریتم IID Spike Detector برای بررسی انحرافات ناگهانی
        var pipeline = _mlContext.Transforms.DetectIidSpike(
            outputColumnName: nameof(AnomalyPrediction.Prediction),
            inputColumnName: nameof(TelemetryDataInput.Value),
            confidence: 98.0,
            pvalueHistoryLength: 30);

        var model = pipeline.Fit(dataView);
        var transformedData = model.Transform(dataView);

        var predictions = _mlContext.Data.CreateEnumerable<AnomalyPrediction>(transformedData, reuseRowObject: false);

        foreach (var p in predictions)
        {
            if (p.Prediction[0] == 1) // 1 نشان‌دهنده آنومالی یا Spike است
            {
                Console.WriteLine($"[ALERT] Anomaly Detected! Value: {p.Prediction[1]}, Score: {p.Prediction[2]}");
            }
        }
    }
}

public class TelemetryDataInput
{
    public float Value { get; set; }
}

public class AnomalyPrediction
{
    // [Alert, RawScore, P-Value]
    public double[] Prediction { get; set; }
}

جمع‌بندی و جدول مقایسه‌ای تکنولوژی‌های پردازش IoT در .NET

برای طراحی یک سیستم جامع تحلیل داده‌های IoT در پلتفرم .NET، انتخاب ابزارهای مناسب برای هر بخش اهمیت فراوانی دارد. جدول زیر راهنمایی سریع برای انتخاب زیرساخت مناسب ارائه می‌دهد:

پلتفرم moderne .NET ابزارها و توانمندی‌های فوق‌العاده‌ای در اختیار مهندسان نرم‌افزار قرار می‌دهد تا سیستم‌هایی مقیاس‌پذیر، کم‌هزینه و پرسرعت برای دریافت و تحلیل داده‌های اینترنت اشیاء بسازند. با ترکیب الگوی Channels برای جداسازی بار، Rx.NET برای تحلیل جریانی، تکنیک‌های Zero-Allocation برای بهینه‌سازی حافظه و ML.NET برای تحلیل‌های هوشمند، می‌توان سامانه‌هایی طراحی کرد که میلیون‌ها رویداد در ثانیه را با پایداری حداکثری پردازش و تحلیل کنند.