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

.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 了. 这会不会极大地影响程序运行性能?

namonai

71 天前

@bthulu 你可以试试 trible buffer ，编号 0 、1 、2 ，读线程实现一个 getIndex()，初始的 valid index 是 0 ，需要对数据更改的时候，index + 1 ，往 1 上写，写完了以后 valide index 也更新到 1 ，这个时候 0 和 1 的数据都是有效的，过了一小段时间，0 就没人访问了。在这段时间里如果又有需要写入的数据，那就往 2 上写。这样子可以始终保证读到的数据是完整的。可以把写入操作放在一个单独的线程里进行，其他线程如果有修改数据的需要，就通过队列传递数据过去。

billccn

71 天前

`devices`这个字典一定程度上就是一个手搓的数据库，你这个里面要考虑的情况很多，比如：

1. 字典需要动态增删吗？不需要的话这个`Concurrent`是徒增开销
2. 字典里面的值（就是每个 Device 实例）会被多个线程同时引用吗？字典的`Concurrent`是不会管里面的值是不是 thread-safe
3. Device 实例需要 referential equivalence 吗？不需要的话建议把这个类变成只读的，每次更新的时候直接替换整个实例最安全
4. Device 与 Device 之间有关系吗？有的话你可能需要考虑如何 atomically 更新这个字典

geelaw

71 天前

>异步方法中根本没办法使用锁, 顶多用用信号量 Semaphore 来代替锁.
>貌似也没法仅对非异步代码进行加锁, 因为同步异步代码是混杂在一块的, 没法单独对非异步代码进行加锁.

规则是 lock 里面不可以有 await （可以实现，但是几乎总是错误的，因此语言层面拒绝这样做），在 async 方法中 lock 是完全 OK 的。

ConcurrentDictionary 已经确保每次访问它的成员都是原子的，然而这不代表对它的访问逻辑就已经线程安全，比如一段代码里连续访问它的成员两次，那么在中途其他线程可能已经修改过了这个字典。说这点是预防针，楼主在 #14 提到这是为了确保 Device 存在 devices 里面的引用安全。

要保证每个 Device 实例线程安全，最简单的思路是细粒度，比如操作每个 device 的时候 lock 之。如果操作过程需要异步，那么我想象中楼主说有多个线程查看 devices 并做一些事，意思是如果 A 线程处理了 device1 则 B 线程应该跳过并处理 device2 ，这种情况下因为 device 被占用时无需等待，所以可以用 interlocked operation 实现：

1. 在 device 上加上一个 int 字段 InUse ，表示目前是否在处理它，初始化为 0 。
2. 要访问一个 device ，先用 Interlocked.Exchange 查看 InUse 并设置为 1 ，如果 InUse 之前也是 1 ，则跳过。
3. 否则 InUse 之前是 0 并且被原子设置为 1 ，此时当前方法认为自己接管该 device 并开始异步硬件 API 操作，在 await 结束、处理完 device 后，重新 Exchange 把 InUse 还原为 0 。

CLMan

71 天前

这个问题的核心是你业务逻辑的“线程安全”是如何定义的（只有你自己知道），至于是用锁、读写锁、Semaphore 、无锁、volatile 等，这些纯粹是实现细节，取决于你对并发相关基础知识（操作系统领域）以及特定语言（这里是.NET ）相关库和语法的熟悉程度。

由于不了解你的业务逻辑实现细节，我只能提问：

- 线程 0 是只写吗，是否依赖 Device 当前的状态？
- 监控线程统计所有设备状态时，以及执行异步操作时，是否允许线程 0 进行更新？
- 监控线程的异步任务与线程 0 是否存在写入相同的内存区域的情况？
- 监控线程的异步任务是否可能执行超过 100 毫秒，如果超过，是否允许多个监控线程的异步任务同时执行？如果允许，它们的写是否冲突？

你至少需要补充以上细节，才能让回答者更好的帮你解决问题。