一个线程更新数据, 多个线程读数据, 这种怎么保证线程安全?

.Net 相关
线程 0 调用硬件异步 API, 拿到数据后, 从 devices 根据 id 取到 Device 实例, 更新硬件最新数据到这个实例上.
同时有多个监控线程每隔 100 毫秒读取一次所有设备状态, 并根据设备状态执行一次或多次耗时较长的异步操作, 并在异步操作执行完成后, 对硬件数据进行部分更新.
这个要怎么做才能确保线程安全?

    // 设备集中存储处
    ConcurrentDictionary<int, Device> devices = new();

    // 设备类
    public class Device
    {
        public int Id { get; init; }
        public bool Enable { get; set; }
        public string Group { get; init; } = "";
        public int[] Locations { get; init; } = Array.Empty<int>();
        public int Margin { get; set; }
        public int RsCount { get; init; }
        public bool EnableSplit { get; init; }
        public int DynamicMerge { get; set; }
        public int Width { get; set; }
        public int Length { get; set; }
        public int LeftLength { get; set; }
        public int LoadEdge { get; set; }
        public int Dest { get; set; }
    }

    // 数据更新线程相关
    public Thread0Executor()
    {
    	public async Task Execute()
        {
            var data = await GetDataFromHardwareApi();
            Update(data, devices);
        }
    }

    // 数据监控处理线程相关
    public MonitorThreadExecutor()
    {
    	public async Task Execute()
        {
            Resolve(devices);
            await Operate0();
            DoSomething();
            await Operate1();
            DoSomething();
        }
        
        public async Task Operate0()
        {
            try
            {
            	await CallApi();
            	Update(devices);
            }
            catch()
            {
            	UpdateIfError(devices);
            }
        }
    }

异步方法中根本没办法使用锁, 顶多用用信号量 Semaphore 来代替锁.

这里也不能对整个 Execute 方法用锁. 因为监控线程中的异步操作耗时是不一定的, 可能因为网络问题花个几分钟都有可能.

貌似也没法仅对非异步代码进行加锁, 因为同步异步代码是混杂在一块的, 没法单独对非异步代码进行加锁.

也考虑过弄个类似 ANDROID 里的 UI 线程和子线程的东西, 数据读取和更新都放在 UI 线程里, 异步操作放在子线程里. 但是搞了半天没搞出来.

最后的最后, 实在没办法了, 我在想要不把 Device 的所有属性都加一个 volatile 关键字. 我这里更新数据的时候基本不会看原来数据是多少, 不会出现count++这种情况, 貌似 volatile 是可行的. 但是实际这个 Device 有几十个属性, 并且有一两千个 Device, 如果每个属性都加一个 volatile 关键字, 那就是 2000*50=100 万个属性带 volatile 了. 这会不会极大地影响程序运行性能?

cloud107202

2024-02-28 11:01:46 +08:00

这里可以考虑做个线程读写分离。没接触过 .Net 我会用 Java 的 type 与 API 描述，自行对应一下:

首先把成员 devices 与相关的操作都封装到一个类型里面，对外暴露一个 public 的阻塞队列成员变量，Java 的话我会用有阻塞语义的 ArrayBlockingQueue. 这个类型在构建的时候(onCreation)，启动一个单线程去 poll 这个 Queue. devices 的更新逻辑都由这个单线程完成

外面的异步操作获取到设备信息后，以 ImmutableEvent 的形式把必要的信息封装描述好，放入队列. 形如 ArrayBlockingQueue<DeviceUpdatedEvent> 这样子，里面的单线程 poll 到事件直接更新 Dictionary 即可。

最后剩下这个“多个监控线程每隔 100 毫秒读取一次所有设备状态” ，这里简单起见可以将 devices 也设置成 public ，直接在外面访问 devices 成员（重点是：一定要约定好，在 poll 的线程之外的逻辑，全部只能 read 这个 ConcurrentDictionary ）。因为 Dictionary 本身使用了线程安全的 ConcurrentDictionary ，对它的 CRUD 是线程安全的，只需要防止外面监控程序获取到某个尚未更新完成的某个 Device 实例（有点像 DB 的脏读），这里给 Device 每个属性设置 volatile 肯定是不合适的：可以考虑前面提到的，在负责 poll 的单线程，获取到更新事件后，不要就地改变 device 对象本身的属性值，而是以 deepCopy 的方式创建个全新的 Device 实例。然后用 ConcurrentDictionary.put(key, value) 的 API 直接更新整个 Device 对象，规避外部监控线程在 scan 的时候，获取到属性更新不完整的 stale state

qping

2024-02-28 14:44:47 +08:00

又仔细看了下，你是多线程写啊，MonitorThread （多个）和 Thread0 都能更新, 那存在一些问题

1. MonitorThread 和 Thread0 是否会写入冲突
如果 MonitorThread 和 Thread0 写入相同得内存，那感觉就是设计有问题
那我假想他们不会冲突

2. 多个 MonitorThread 冲突的问题
多个 MonitorThread 每次都更新全部的 devices ，这个设计也很奇怪

假设已经做到通过锁或其他手段，保证一个 MonitorThread 更新是原子级别的。
MonitorThread A 先启动，MonitorThread B 后启动，因为等待时间长 A 的结果却比 B 后写入，这样没有问题吗？

我觉得，应该可以有多个 MonitorThread 线程，但是每个 Device 只能同一时间被一个 MonitorThread 更新
实现方法上，可以用队列，每次更新 MonitorThread 从队列中取一个 Device ，如果更新完重新还回