在 .NET 环境下实现跨进程高频率读写数据

news2025/5/25 7:42:02

✅ 技术选型说明

📦 示例场景

🧩 数据结构定义

🚦 核心同步机制

🧑‍💻 消费者实现

⚡ 性能优化技巧

🛠 部署注意事项

📈 性能基准（理论值）

在 .NET 环境下实现跨进程高频率读写数据，通常需要结合高性能通信机制（如共享内存、命名管道或内存映射文件）和线程同步技术。以下是基于 内存映射文件（Memory-Mapped Files） 和 信号量同步 的完整案例，适合高频数据传输场景（如实时传感器数据采集、高频交易日志等）。

技术选型说明

MemoryMappedFile
- 共享内存机制，适合跨进程高速读写
- 通过 MemoryMappedViewAccessor 操作二进制数据
SemaphoreSlim
- 轻量级同步原语，避免忙等待
环形缓冲区（Ring Buffer）
- 高效管理高频数据流，减少锁竞争

示例场景

模拟 传感器数据采集系统：

进程 A：每秒生成 100,000 条传感器数据（包含时间戳和数值）
进程 B：实时消费数据并计算移动平均值

数据结构定义

// 通用数据结构（两个进程需共享定义）
[StructLayout(LayoutKind.Sequential, Pack = 1)]
public struct SensorData
{
    public long Timestamp;      // 时间戳（毫秒）
    public float Value;         // 传感器数值
}

核心同步机制

使用 MemoryMappedFile + Semaphore 实现无锁协作：

// 共享内存布局
const int BufferSize = 1024 * 1024; // 1MB 缓冲区
const string MapName = "SensorDataBuffer";
const string WriteSemaphore = "SensorWriteSemaphore";
const string ReadSemaphore = "SensorReadSemaphore";

// 生产者写入逻辑
using var mmf = MemoryMappedFile.CreateOrOpen(MapName, BufferSize);
using var writerSem = new SemaphoreSlim(1, 1); // 写信号量
using var readerSem = new SemaphoreSlim(0, 1); // 读信号量

var accessor = mmf.CreateViewAccessor();
int writeIndex = 0;

for (int i = 0; i < 1000000; i++)
{
    await writerSem.WaitAsync(); // 等待写权限

    var data = new SensorData
    {
        Timestamp = DateTime.Now.Ticks,
        Value = (float)(Math.Sin(i * 0.01) * 100)
    };

    accessor.Write(writeIndex, ref data);
    writeIndex += Marshal.SizeOf<SensorData>();

    if (writeIndex >= BufferSize)
        writeIndex = 0; // 环形缓冲区

    readerSem.Release(); // 通知消费者
}

消费者实现

// 消费者读取逻辑
using var mmf = MemoryMappedFile.OpenExisting(MapName);
using var writerSem = new SemaphoreSlim(1, 1);
using var readerSem = new SemaphoreSlim(0, 1);

var accessor = mmf.CreateViewAccessor();
int readIndex = 0;
float sum = 0;
int count = 0;

while (true)
{
    await readerSem.WaitAsync(); // 等待数据

    var data = new SensorData();
    accessor.Read(readIndex, out data);
    readIndex += Marshal.SizeOf<SensorData>();

    if (readIndex >= BufferSize)
        readIndex = 0;

    writerSem.Release(); // 归还写权限

    // 计算移动平均（示例）
    sum += data.Value;
    count++;
    if (count % 100 == 0)
    {
        Console.WriteLine($"Moving Avg: {sum / 100:F2}");
        sum = 0;
        count = 0;
    }
}