C# 实现条件变量

C# 进程通信系列

第一章共享内存
第二章条件变量（本章）
第三章消息队列

文章目录

C# 进程通信系列
前言
一、关键实现
- 1、用到的主要对象
- 2、初始化区分创建和打开
- 3、变量放到共享内存
- 4、等待和释放逻辑
二、完整代码
三、使用示例
- 1、线程同步控制
- 2、进程同步控制
总结

前言

C#提供的多进程同步对象有互斥锁和信号量，但是并没有条件变量。虽然信号量条件变量一定程度可以等效，但是具体的使用还是会有区别。比如说消息队列用条件变量就比信号量方便，用信号量要么缺乏灵活性，要么辅助代码已经和实现一个条件变量没区别了。本文提供了一种条件变量的实现方法，可以用于进程间的同步控制。

一、关键实现

1、用到的主要对象

下列对象都是跨进程的

//互斥变量，
Mutex _mtx;
//等待发送信号量
Semaphore _waitSem;
//等待完成信号量
Semaphore _waitDone;
//共享内存，用于存储计数变量
MemoryMappedFile _mmf;
//共享内存的读写对象
MemoryMappedViewAccessor _mmva;

2、初始化区分创建和打开

利用Mutex判断是创建还是打开

bool isCreateNew;
_mtx = new Mutex(false, name, out isCreateNew);if(isCreateNew){//只能在创建时，初始化共享变量}

3、变量放到共享内存

条件变量需要的计算对象就两个Waiting、Signals表示等待数和释放数。

//放到共享内存的数据
struct SharedData
{public int Waiting;public int Signals;
}
SharedData Data
{set{_mmva.Write(0, ref value);}get{SharedData ret;_mmva.Read(0, out ret);return ret;}
}

4、等待和释放逻辑

参考了SDL2的条件变量实现，具体略。有兴趣的朋友可以自行查找源码查看。

二、完整代码

using System.IO.MemoryMappedFiles;
using System.Runtime.InteropServices;namespace AC
{/************************************************************************* @Project:  	AC::ConditionVariable* @Decription:   条件变量*                支持跨进程* @Verision:  	v1.0.0    *              更新日志*              v1.0.0：实现基本功能*              v1.1.0:优化进程内逻辑* @Author:  	Xin* @Create:  	2024/07/18 15:25:00* @LastUpdate:  2024/07/21 20:53:00************************************************************************* Copyright @ 2025. All rights reserved.************************************************************************/class ConditionVariable : IDisposable{/// <summary>/// 构造方法/// 本进程内使用条件变量，会更轻量一些。/// </summary>public ConditionVariable(){bool isCreateNew;Initialize(null, out isCreateNew);}/// <summary>/// 构造方法/// </summary>/// <param name="name">唯一名称，系统级别，不同进程创建相同名称的本对象，就是同一个条件变量。</param>public ConditionVariable(string name){bool isCreateNew;Initialize(name, out isCreateNew);}/// <summary>/// 构造方法/// </summary>/// <param name="name">唯一名称，系统级别，不同进程创建相同名称的本对象，就是同一个条件变量。</param>/// <param name="isCreateNew">表示是否新创建，是则是创建，否则是打开已存在的。</param>public ConditionVariable(string? name, out bool isCreateNew){Initialize(name, out isCreateNew);}/// <summary>/// 等待/// </summary>/// <param name="outerMtx">外部锁</param>public void WaitOne(Mutex outerMtx){WaitOne(Timeout.InfiniteTimeSpan, outerMtx);}/// <summary>/// 等待超时/// </summary>/// <param name="timeout">超时时间</param>/// <param name="outerMtx">外部锁，可以跨进程使用</param>/// <returns>是则成功，否则超时</returns>public bool WaitOne(TimeSpan timeout, Mutex outerMtx){return WaitOne(timeout, (object)outerMtx);}/// <summary>/// 等待/// </summary>/// <param name="outerMtx">外部锁,为lock关键字的对象，只能本进程使用</param>public void WaitOne(object outerMtx){WaitOne(Timeout.InfiniteTimeSpan, outerMtx);}/// <summary>/// 等待超时/// </summary>/// <param name="timeout">超时时间</param>/// <param name="outerMtx">外部锁</param>/// <returns>是则成功，否则超时</returns>public bool WaitOne(TimeSpan timeout, object outerMtx){bool isNotTimeout;//记录等待数量if (_mtx != null) _mtx.WaitOne(); else Monitor.Enter(_waitSem);var ws = Data;ws.Waiting++;Data = ws;if (_mtx != null) _mtx.ReleaseMutex(); else Monitor.Exit(_waitSem);//解除外部的互斥锁，让其他线程可以进入条件等待。if (outerMtx is Mutex) ((Mutex)outerMtx).ReleaseMutex(); else Monitor.Exit(outerMtx);//等待信号isNotTimeout = _waitSem.WaitOne(timeout);if (_mtx != null) _mtx.WaitOne(); else Monitor.Enter(_waitSem);ws = Data;if (isNotTimeout && ws.Signals > 0){//通知发送信号的线程，等待完成。_waitDone.Release();ws.Signals--;}ws.Waiting--;Data = ws;if (_mtx != null) _mtx.ReleaseMutex(); else Monitor.Exit(_waitSem);//加上外部互斥锁，还原外部的锁状态。if (outerMtx is Mutex) ((Mutex)outerMtx).WaitOne(); else Monitor.Enter(outerMtx);return !isNotTimeout;}/// <summary>/// 释放，通知/// </summary>public void Release(){if (_mtx != null) _mtx.WaitOne(); else Monitor.Enter(_waitSem);var ws = Data;if (ws.Waiting > ws.Signals){ws.Signals++;Data = ws;_waitSem.Release();if (_mtx != null) _mtx.ReleaseMutex(); else Monitor.Exit(_waitSem);_waitDone.WaitOne();}else{if (_mtx != null) _mtx.ReleaseMutex(); else Monitor.Exit(_waitSem);}}/// <summary>/// 释放全部，广播/// </summary>public void ReleaseAll(){if (_mtx != null) _mtx.WaitOne(); else Monitor.Enter(_waitSem);var ws = Data;if (ws.Waiting > ws.Signals){int waiting = ws.Waiting - ws.Signals;ws.Signals = ws.Waiting;Data = ws;_waitSem.Release(waiting);if (_mtx != null) _mtx.ReleaseMutex(); else Monitor.Exit(_waitSem);_waitDone.WaitOne(waiting);}else{if (_mtx != null) _mtx.ReleaseMutex(); else Monitor.Exit(_waitSem); _mtx.ReleaseMutex();}}/// <summary>/// 销毁对象，只会销毁当前实例，如果多个打开同个名称，其他对象不受影响/// </summary>public void Dispose(){_mtx?.Dispose();_waitSem.Dispose();_waitDone.Dispose();_mmva?.Dispose();_mmf?.Dispose();}void Initialize(string? name, out bool isCreateNew){Mutex? mtx = null;Semaphore? waitSem = null;Semaphore? waitDone = null;MemoryMappedFile? mmf = null;MemoryMappedViewAccessor? mmva = null;try{if (name != null){mtx = _mtx = new Mutex(false, name, out isCreateNew);_mtx.WaitOne();try{waitSem = _waitSem = new Semaphore(0, int.MaxValue, name + ".cv.ws");waitDone = _waitDone = new Semaphore(0, int.MaxValue, name + ".cv.wd");var _shmPath = Path.Combine(_TempDirectory, name + ".cv");mmf = _mmf = MemoryMappedFile.CreateFromFile(File.Open(_shmPath, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.ReadWrite), null, Marshal.SizeOf<SharedData>(), MemoryMappedFileAccess.ReadWrite, HandleInheritability.Inheritable, false);mmva = _mmva = _mmf.CreateViewAccessor();if (isCreateNew) Data = new SharedData() { Signals = 0, Waiting = 0 };}finally{_mtx.ReleaseMutex();}}else{waitSem = _waitSem = new Semaphore(0, int.MaxValue);waitDone = _waitDone = new Semaphore(0, int.MaxValue);isCreateNew = true;Data = new SharedData() { Signals = 0, Waiting = 0 };}}catch{mtx?.Dispose();waitSem?.Dispose();waitDone?.Dispose();mmf?.Dispose();mmva?.Dispose();isCreateNew = false;throw;}}Mutex? _mtx;Semaphore _waitSem;Semaphore _waitDone;MemoryMappedFile ?_mmf;MemoryMappedViewAccessor ?_mmva;SharedData _data;struct SharedData{public int Waiting;public int Signals;}SharedData Data{set{if (_mmva != null){_mmva.Write(0, ref value);}else{_data = value;}}get{if (_mmva != null){SharedData ret;_mmva.Read(0, out ret);return ret;}return _data;}}static string _TempDirectory = Path.GetTempPath() + "EE3E9111-8F65-4D68-AB2B-A018DD9ECF3C";}
}

三、使用示例

1、线程同步控制

using AC;
ConditionVariable cv = new ConditionVariable();
string text = "";
//子线程发送消息
new Thread(() =>
{int n = 0;while (true){lock(cv){text = (n++).ToString();//通知主线程cv.Release();}}}).Start();
//主线程接收消息
while (true)
{lock(cv){//等待子消息cv.WaitOne(cv);Console.WriteLine(text);}
}

在这里插入图片描述

2、进程同步控制

进程A

//不同进程名称相同就是同一个对象
ConditionVariable cv = new ConditionVariable("cv1");
Mutex mutex = new Mutex(false,"mx1");
//进程A发送消息
while (true)
{mutex.WaitOne();//共享内存写略//通知进程Bcv.Release();mutex.ReleaseMutex();
}

进程B

//不同进程名称相同就是同一个对象
ConditionVariable cv = new ConditionVariable("cv1");
Mutex mutex = new Mutex(false,"mx1");
//进程B接收消息
while (true)
{mutex.WaitOne();//等待进A程消息cv.WaitOne(mutex);//共享内存读略Console.WriteLine("收到进程A消息");mutex.ReleaseMutex();
}