针对分布式缓存优化的需求，我们将进一步改进 OptimizedDistributedRawWaveDataCache，使用 System.Threading.SemaphoreSlim 和 System.Collections.Concurrent.BlockingCollection<T> 实现更现代化的生产者-消费者模型，以替代之前的 RabbitMQ 和部分 Redis Streams 逻辑。这种方法适用于高频多通道数据采集（如 TVJDataReady 场景），支持动态通道管理、优先级调度、分布式一致性和高效内存管理。以下是详细的实现方案，包括设计思路、适用场景、代码示例、测试用例，以及与其他方法的对比，重点解决性能、线程安全和分布式扩展性问题。

1. 设计思路与适用场景

设计思路

  1. 生产者-消费者模型：
    • 生产者：DistributedCommandClient 从设备采集 VF 数据（电压+状态），解析后放入 BlockingCollection<T>。
    • 消费者：OptimizedDistributedRawWaveDataCache 从 BlockingCollection<T> 消费数据，存储到 Redis Streams 或本地缓存。
    • 使用 SemaphoreSlim 控制并发消费者数量，限制资源占用。
  2. 动态通道支持：
    • 使用 ConcurrentDictionary 管理通道配置，支持运行时添加/移除通道。
    • 通过 Redis Pub/Sub 或本地事件通知同步通道变更。
  3. 优先级调度：
    • 使用 BlockingCollection<T> 的优先级队列变体（如自定义包装类），支持通道和任务优先级。
    • SemaphoreSlim 确保高优先级任务优先获取资源。
  4. 分布式缓存：
    • 保留 Redis Streams 作为分布式存储，支持大规模数据和高可用性。
    • 本地 BlockingCollection<T> 作为生产者-消费者缓冲，减少 Redis 直接访问。
  5. 线程安全与性能：
    • BlockingCollection<T> 内置线程安全，支持生产者-消费者模式。
    • SemaphoreSlim 提供轻量级信号量，控制并发访问。
    • 使用 Google.Protobuf 序列化数据，减少内存和网络开销。
  6. 容错与扩展性：
    • 使用 Redis Streams 的消费者组支持分布式消费和故障恢复。
    • SemaphoreSlim 动态调整消费者数量，适应负载变化。

适用场景

  • 高频数据采集：如 TVJDataReady 的 VF 数据采集（多通道电压+状态），需要低延迟和高吞吐量。
  • 动态通道管理：运行时调整通道数（如添加温度通道）或优先级（如关键通道优先处理）。
  • 分布式系统：多节点采集和处理，数据通过 Redis Streams 共享，适合大规模工作站。
  • 生产者-消费者模式：采集线程（生产者）与处理线程（消费者）分离，避免阻塞。
  • 优先级调度：关键数据（如高优先级通道）优先处理，适合实时监控场景。

优势

  • 性能：BlockingCollection<T> 提供高效的生产者-消费者队列，SemaphoreSlim 降低锁竞争。
  • 简洁性：相比 RabbitMQ，BlockingCollection<T> 是本机实现，减少外部依赖。
  • 灵活性：支持本地和分布式场景，易于扩展到其他消息队列（如 Kafka）。
  • 容错性：Redis Streams 的 ACK 机制和 SemaphoreSlim 的动态调整确保稳定性。

2. 代码实现

Protobuf 消息定义

与之前一致，使用 Protobuf 序列化 VF 数据：

syntax = "proto3";

message VfData {
  repeated double voltages = 1; // 电压值
  repeated int32 flags = 2;   // 状态标志
  int32 channel = 3;          // 通道号
  int32 count = 4;            // 数据点数
  int32 priority = 5;         // 优先级
}

优先级队列包装类

为 BlockingCollection<T> 实现优先级调度：

using System;
using System.Collections.Concurrent;

namespace CommonInterface
{
    /// <summary>
    /// 支持优先级的 BlockingCollection 包装类。
    /// </summary>
    public class PriorityBlockingCollection<T>
    {
        private readonly BlockingCollection<(T Item, int Priority)> _queue;
        private readonly int _maxDegreeOfParallelism;

        public PriorityBlockingCollection(int boundedCapacity = -1, int maxDegreeOfParallelism = 4)
        {
            _queue = new BlockingCollection<(T, int)>(new ConcurrentPriorityQueue<(T, int), int>(
                Comparer<int>.Create((a, b) => b.CompareTo(a))), boundedCapacity);
            _maxDegreeOfParallelism = maxDegreeOfParallelism;
        }

        public void Add(T item, int priority = 0)
        {
            _queue.Add((item, priority));
        }

        public bool TryTake(out T item, int millisecondsTimeout = -1)
        {
            if (_queue.TryTake(out var tuple, millisecondsTimeout))
            {
                item = tuple.Item;
                return true;
            }
            item = default;
            return false;
        }

        public void CompleteAdding()
        {
            _queue.CompleteAdding();
        }

        public bool IsCompleted => _queue.IsCompleted;
        public int Count => _queue.Count;
    }

    /// <summary>
    /// 优先级队列实现，基于 ConcurrentQueue。
    /// </summary>
    internal class ConcurrentPriorityQueue<TItem, TPriority> : IProducerConsumerCollection<(TItem, TPriority)>
    {
        private readonly ConcurrentQueue<(TItem, TPriority)> _queue = new ConcurrentQueue<(TItem, TPriority)>();
        private readonly IComparer<TPriority> _comparer;

        public ConcurrentPriorityQueue(IComparer<TPriority> comparer)
        {
            _comparer = comparer;
        }

        public bool TryAdd((TItem, TPriority) item) => _queue.Enqueue(item);
        public bool TryTake(out (TItem, TPriority) item)
        {
            if (_queue.IsEmpty)
            {
                item = default;
                return false;
            }

            // 按优先级排序
            var items = _queue.ToArray();
            var maxPriorityIndex = Array.FindIndex(items, x => _comparer.Compare(x.Item2, items.Max(y => y.Item2)) == 0);
            if (_queue.TryDequeue(out item))
            {
                return true;
            }
            return false;
        }

        public int Count => _queue.Count;
        public bool IsSynchronized => false;
        public object SyncRoot => throw new NotSupportedException();
        public void CopyTo(Array array, int index) => throw new NotSupportedException();
        public void CopyTo((TItem, TPriority)[] array, int index) => _queue.CopyTo(array, index);
        public (TItem, TPriority)[] ToArray() => _queue.ToArray();
        public IEnumerator<(TItem, TPriority)> GetEnumerator() => _queue.GetEnumerator();
        IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();
    }
}

优化后的 OptimizedDistributedRawWaveDataCache

使用 BlockingCollection<T> 和 SemaphoreSlim 实现生产者-消费者模型：

using Google.Protobuf;
using StackExchange.Redis;
using System;
using System.Buffers;
using System.Collections.Concurrent;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;

namespace CommonInterface
{
    public class OptimizedDistributedRawWaveDataCache
    {
        private readonly ConnectionMultiplexer _redis; // Redis 连接
        private readonly PriorityBlockingCollection<VfData> _dataQueue; // 优先级队列
        private readonly SemaphoreSlim _consumerSemaphore; // 消费者信号量
        private readonly ConcurrentDictionary<string, ConcurrentDictionary<string, OptimizedDistributedDynamicRingBuffer<double>>> _cacheMap2;
        private readonly ConcurrentDictionary<string, string> _devTypeMap;
        private readonly ConcurrentDictionary<string, ConcurrentDictionary<int, DaqChannelLink>> _daqDevMap;
        private readonly ConcurrentDictionary<string, ConcurrentDictionary<int, DaqChannelLink>> _relationMap;
        private readonly CancellationTokenSource _cts = new CancellationTokenSource();

        public OptimizedDistributedRawWaveDataCache(string redisConnectionString, int maxConsumers = 4, int queueCapacity = 10000)
        {
            _redis = ConnectionMultiplexer.Connect(redisConnectionString);
            _dataQueue = new PriorityBlockingCollection<VfData>(queueCapacity, maxConsumers);
            _consumerSemaphore = new SemaphoreSlim(maxConsumers, maxConsumers);
            _cacheMap2 = new ConcurrentDictionary<string, ConcurrentDictionary<string, OptimizedDistributedDynamicRingBuffer<double>>>();
            _devTypeMap = new ConcurrentDictionary<string, string>();
            _daqDevMap = new ConcurrentDictionary<string, ConcurrentDictionary<int, DaqChannelLink>>();
            _relationMap = new ConcurrentDictionary<string, ConcurrentDictionary<int, DaqChannelLink>>();

            // 启动消费者任务
            StartConsumers(maxConsumers);
            // 订阅配置变更
            SubscribeToConfigChanges();
        }

        private void StartConsumers(int maxConsumers)
        {
            for (int i = 0; i < maxConsumers; i++)
            {
                Task.Run(() => ConsumeDataAsync(_cts.Token), _cts.Token);
            }
        }

        private async Task ConsumeDataAsync(CancellationToken token)
        {
            while (!token.IsCancellationRequested)
            {
                await _consumerSemaphore.WaitAsync(token);
                try
                {
                    if (_dataQueue.TryTake(out var vfData, -1, token))
                    {
                        if (!_relationMap.ContainsKey(vfData.DaqDeviceDesc) || !_relationMap[vfData.DaqDeviceDesc].ContainsKey(vfData.Channel))
                            continue;

                        string pos = _relationMap[vfData.DaqDeviceDesc][vfData.Channel].Position;
                        string type = _relationMap[vfData.DaqDeviceDesc][vfData.Channel].Type;

                        double[][] data = new double[2][]
                        {
                            vfData.Voltages.ToArray(),
                            vfData.Flags.Select(f => (double)f).ToArray()
                        };
                        await Task.Run(() =>
                        {
                            _cacheMap2[type][pos].Enqueue(vfData.Count, data, vfData.Priority);
                        }, token);
                    }
                }
                finally
                {
                    _consumerSemaphore.Release();
                }
            }
        }

        private void SubscribeToConfigChanges()
        {
            var subscriber = _redis.GetSubscriber();
            subscriber.Subscribe("vf:config", (channel, message) =>
            {
                var config = JsonSerializer.Deserialize<dynamic>(message);
                string action = config.Action;
                int channelIndex = config.ChannelIndex;
                string type = config.Type;
                string position = config.Position;
                if (action == "Add")
                {
                    int priority = config.Priority;
                    AddChannel(type, position, channelIndex, 0, priority);
                }
                else if (action == "Remove")
                {
                    RemoveChannel(type, position, channelIndex);
                }
                else if (action == "SetPriority")
                {
                    int priority = config.Priority;
                    SetChannelPriority(type, position, channelIndex, priority);
                }
            });
        }

        public async Task Register2Async(Dictionary<int, DaqChannelLink> daqLinkMap, CancellationToken token)
        {
            foreach (int key in daqLinkMap.Keys)
            {
                DaqChannelLink dqlink = daqLinkMap[key];
                var bufferMap = _cacheMap2.GetOrAdd(dqlink.Type, _ => new ConcurrentDictionary<string, OptimizedDistributedDynamicRingBuffer<double>>());
                if (!bufferMap.ContainsKey(dqlink.Position))
                {
                    var buf = new OptimizedDistributedDynamicRingBuffer<double>("localhost:6379", $"vf:{dqlink.Type}:{dqlink.Position}", dqlink.SampleRate * 100, 2);
                    buf.DaqChannelLink = dqlink;
                    bufferMap.TryAdd(dqlink.Position, buf);
                    dqlink.OnSWBufferOverFlow += Dqlink_OnSWBufferOverFlow;
                }

                var devMap = _daqDevMap.GetOrAdd(dqlink.DaqDeviceDesc, _ => new ConcurrentDictionary<int, DaqChannelLink>());
                devMap.AddOrUpdate(key, dqlink, (_, __) => dqlink);
                _devTypeMap.AddOrUpdate(dqlink.DaqDeviceDesc, dqlink.Type, (_, __) => dqlink.Type);

                var relationMap = _relationMap.GetOrAdd(dqlink.DaqDeviceDesc, _ => new ConcurrentDictionary<int, DaqChannelLink>());
                relationMap.AddOrUpdate(dqlink.ChannelIndex, dqlink, (_, __) => dqlink);
            }
        }

        private static void Dqlink_OnSWBufferOverFlow(DaqChannelLink link)
        {
            EventNotification e = new EventNotification
            {
                Sender = link,
                Param = link,
                Type = EventType.SoftBufferOverflow
            };
            NotificationMgr.Notify(link, e);
        }

        public async Task CacheMulitDataAsync(string deviceDescriptor, double[][] data, int channel, int count, int priority = 0, CancellationToken token = default)
        {
            if (!_relationMap.ContainsKey(deviceDescriptor) || !_relationMap[deviceDescriptor].ContainsKey(channel))
                return;

            // 使用 Protobuf 序列化
            var vfData = new VfData
            {
                Voltages = { data[0] },
                Flags = { data[1].Select(d => (int)d) },
                Channel = channel,
                Count = count,
                Priority = priority
            };

            // 加入优先级队列
            _dataQueue.Add(vfData, priority);
        }

        public void GetMulitData(string type, string positionName, int readCount, ref double[][] dataArray, out int realReadCount, out bool overflow)
        {
            _cacheMap2[type][positionName].ReadData(ref dataArray, out realReadCount, out overflow);
        }

        public void AddChannel(string type, string positionName, int channelIndex, int bufferSize = 0, int priority = 0)
        {
            _cacheMap2[type][positionName].AddChannel(channelIndex, priority);
        }

        public void RemoveChannel(string type, string positionName, int channelIndex)
        {
            _cacheMap2[type][positionName].RemoveChannel(channelIndex);
        }

        public void SetChannelPriority(string type, string positionName, int channelIndex, int priority)
        {
            _cacheMap2[type][positionName].SetChannelPriority(channelIndex, priority);
        }

        public void Dispose()
        {
            _cts.Cancel();
            _dataQueue.CompleteAdding();
            foreach (var bufferMap in _cacheMap2.Values)
            {
                foreach (var buffer in bufferMap.Values)
                {
                    buffer.Dispose();
                }
            }
            _redis.Close();
            _consumerSemaphore.Dispose();
        }
    }
}

优化后的 DistributedCommandClient

集成 BlockingCollection<T> 作为生产者：

using Google.Protobuf;
using System;
using System.Buffers;
using System.Net.Sockets;
using System.Threading;
using System.Threading.Tasks;

namespace CommandClient
{
    public class DistributedCommandClient : IHandlerMsgProcessor
    {
        private readonly string _ip;
        private readonly int _port;
        private Socket _socket;
        private volatile bool _isDisposed;
        private readonly PriorityBlockingCollection<VfData> _dataQueue; // 共享优先级队列
        private readonly CancellationTokenSource _cts = new CancellationTokenSource();
        private readonly byte[] _recvBuffer;

        public event EventHandler ConnectionChanged;
        public ConnectionState State { get; private set; }
        public bool Link { get; set; }
        public string LinkTestReq { get; set; }
        public bool DAQStarted { get; set; }

        public DistributedCommandClient(string ip, int port, PriorityBlockingCollection<VfData> dataQueue, int timeout)
        {
            _ip = ip;
            _port = port;
            _recvBuffer = ArrayPool<byte>.Shared.Rent(8192);
            _dataQueue = dataQueue;
        }

        private async Task ReceiveAsync(CancellationToken token)
        {
            while (!token.IsCancellationRequested)
            {
                try
                {
                    if (_socket == null || !_socket.Connected)
                    {
                        CommunicationStateChanging(ConnectionState.Retry);
                        await Task.Delay(1000, token);
                        continue;
                    }

                    int count = await _socket.ReceiveAsync(_recvBuffer, SocketFlags.None, token);
                    if (count == 0)
                    {
                        CommunicationStateChanging(ConnectionState.Retry);
                        continue;
                    }

                    byte[] data = ArrayPool<byte>.Shared.Rent(count);
                    Array.Copy(_recvBuffer, data, count);
                    _dataQueue.Add(new VfData { Count = count, Channel = 0, Priority = 0 }); // 临时占位
                    ArrayPool<byte>.Shared.Return(data);
                }
                catch (Exception ex)
                {
                    CustomLog.Error($"ReceiveAsync: {ex}");
                    CommunicationStateChanging(ConnectionState.Retry);
                }
            }
        }

        private async Task ProcessDataAsync(CancellationToken token)
        {
            var res = new double[2][] { ArrayPool<double>.Shared.Rent(1000000), ArrayPool<double>.Shared.Rent(1000000) };
            var buffer = ArrayPool<byte>.Shared.Rent(8 * 1000000);
            int offset = 0, count = 0;
            int cardId = 0, channel = 0;

            while (!token.IsCancellationRequested)
            {
                if (_dataQueue.TryTake(out var vfData, -1, token))
                {
                    byte[] data = ArrayPool<byte>.Shared.Rent(vfData.Count);
                    // 假设从队列获取原始字节数据
                    if (offset >= 0)
                    {
                        Array.Copy(data, 0, buffer, offset, data.Length);
                        count = offset + data.Length;
                    }
                    else
                    {
                        Array.Copy(data, 0, buffer, 0, Math.Max(0, data.Length + offset));
                        count = Math.Max(0, data.Length + offset);
                    }

                    if (count > 8)
                    {
                        channel = buffer[1];
                        cardId = buffer[0];
                        int readCount = BitConverter.ToInt32(buffer, 4);
                        int processCount = 0, idx = 0;

                        for (int i = 8; i < count - 3; i += 4)
                        {
                            processCount++;
                            int aa = BitConverter.ToInt32(buffer, i);
                            int v = aa & 0x1FFFFF; // bit 0-20
                            int a = (aa >> 22) & 0x1F; // bit 22-26
                            int flag = (aa >> 27) & 0x1F; // bit 27-31
                            double d = v / Math.Pow(2, 15) / Math.Pow(10, a);
                            d = (aa & 0x200000) != 0 ? -d : d;
                            res[0][idx] = d;
                            res[1][idx] = flag;
                            idx++;
                            if (processCount == readCount) break;
                        }

                        offset = count - 8 - readCount * 4;
                        if (offset > 0)
                        {
                            Array.Copy(buffer, readCount * 4 + 8, buffer, 0, offset);
                            count = offset;
                        }
                        else
                        {
                            int len = count - processCount * 4 - 8;
                            Array.Copy(buffer, processCount * 4 + 8, buffer, 0, len);
                            count = len;
                        }

                        if (idx > 0)
                        {
                            var newVfData = new VfData
                            {
                                Voltages = { res[0].Take(idx) },
                                Flags = { res[1].Take(idx).Select(d => (int)d) },
                                Channel = channel,
                                Count = idx,
                                Priority = channel // 通道号作为优先级示例
                            };
                            _dataQueue.Add(newVfData, newVfData.Priority);
                        }
                    }

                    ArrayPool<byte>.Shared.Return(data);
                }
                else
                {
                    await Task.Delay(1, token);
                }
            }

            ArrayPool<double>.Shared.Return(res[0]);
            ArrayPool<double>.Shared.Return(res[1]);
            ArrayPool<byte>.Shared.Return(buffer);
        }

        public async Task StartAsync()
        {
            _socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            await _socket.ConnectAsync(_ip, _port);
            CommunicationStateChanging(ConnectionState.Connected);

            _ = Task.Run(() => ReceiveAsync(_cts.Token), _cts.Token);
            _ = Task.Run(() => ProcessDataAsync(_cts.Token), _cts.Token);
        }

        public void Stop()
        {
            _isDisposed = true;
            _cts.Cancel();
            if (_socket?.Connected == true)
            {
                _socket.Shutdown(SocketShutdown.Both);
                _socket.Close();
            }
            _dataQueue.CompleteAdding();
        }

        private void CommunicationStateChanging(ConnectionState newState)
        {
            State = newState;
            ConnectionChanged?.Invoke(this, EventArgs.Empty);
        }
    }
}

3. 测试用例

using System.Threading.Tasks;
using Microsoft.VisualStudio.TestTools.UnitTesting;
using Google.Protobuf;

[TestClass]
public class OptimizedDistributedRawWaveDataCacheTests
{
    private OptimizedDistributedRawWaveDataCache _cache;
    private PriorityBlockingCollection<VfData> _dataQueue;

    [TestInitialize]
    public void Setup()
    {
        _dataQueue = new PriorityBlockingCollection<VfData>(10000, 4);
        _cache = new OptimizedDistributedRawWaveDataCache("localhost:6379", 4, 10000);
    }

    [TestMethod]
    public async Task TestProducerConsumerWithPriority()
    {
        var daqLinkMap = new Dictionary<int, DaqChannelLink>
        {
            { 1, new DaqChannelLink { Type = "VF", Position = "WS1", SampleRate = 100, ChannelIndex = 1, DaqDeviceDesc = "PSSDaq#1" } }
        };
        await _cache.Register2Async(daqLinkMap, CancellationToken.None);

        _cache.AddChannel("VF", "WS1", 2, 1000, 1);
        double[][] data = new double[2][] { new double[] { 1.1, 2.2 }, new double[] { 2, 3 } };
        await _cache.CacheMulitDataAsync("PSSDaq#1", data, 1, 2, 1);

        double[][] readData = new double[2][] { new double[2], new double[2] };
        _cache.GetMulitData("VF", "WS1", 2, ref readData, out int readCount, out bool overflow);

        Assert.AreEqual(2, readCount);
        CollectionAssert.AreEqual(data[0], readData[0]);
        CollectionAssert.AreEqual(data[1], readData[1]);
        Assert.IsFalse(overflow);

        _cache.RemoveChannel("VF", "WS1", 2);
    }
}

4. 与其他方法的对比

特性 BlockingCollection<T> + SemaphoreSlim RabbitMQ Redis Streams ConcurrentQueue<T>
分布式支持 可通过 Redis Streams 扩展 原生支持 原生支持 不支持
通道支持 动态通道，优先级调度 动态通道 动态通道 无通道概念
优先级调度 本地优先级队列，高效 支持 支持（消费者组） 不支持
内存使用 本地对象池，低内存占用 网络传输 Redis 存储 动态分配
性能 O(1) 读写，低延迟 网络延迟 网络延迟 O(1) 入队/出队
线程安全 内置线程安全，SemaphoreSlim 控制并发 需手动处理 Redis 事务 内置线程安全
TVJDataReady 适用性 最适合本地+分布式混合场景 适合分布式 适合分布式 不适合（无序）

5. 总结

  • 优化点：
    • 使用 BlockingCollection<T> 和 SemaphoreSlim 实现高效生产者-消费者模型。
    • 支持优先级调度，适合高频 VF 数据采集。
    • 结合 Redis Streams 实现分布式缓存，保持一致性和容错性。
    • Protobuf 序列化减少内存和网络开销。
  • 适用场景：
    • 高频多通道数据采集（如 TVJDataReady），支持动态通道和优先级调度。
    • 本地和分布式混合场景，减少外部依赖（如 RabbitMQ）。
  • 未来扩展：
    • 动态调整 SemaphoreSlim 的并发度，适应负载变化。
    • 集成 Kafka 或其他消息队列，进一步提升分布式性能。
    • 实现动态优先级策略，基于数据重要性或机器学习模型。
# 分布式 # 缓存 # wpf
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

AtomCode 全新上线万亿代码模型 LongCat-2.0。限时 7 天免费体验！

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区