跨进程使用.NET Channel<T>：是否存在可行实现方案？

原生的.NET Channel API是进程内专属的通信机制，本身不支持跨进程访问——它的底层依赖进程内的内存结构和同步原语，没有提供命名寻址或跨进程共享的能力。但可以通过以下两种方式实现类似Channel的跨进程生产/消费逻辑：

方案1：基于共享内存+命名同步原语手动封装 #

借助.NET的MemoryMappedFile实现跨进程共享内存，配合Mutex、Semaphore这类命名同步原语（同一机器内的进程可通过名称访问），手动模拟Channel的生产/消费行为。

生产者示例代码 #

// 创建或打开命名内存映射文件，分配1MB内存空间
using var mmf = MemoryMappedFile.CreateOrOpen("CrossProcessMsgChannel", 1024 * 1024);
// 创建命名互斥体，用于同步内存访问
using var mutex = new Mutex(false, "CrossProcessMsgMutex");
// 创建命名信号量，控制消费端的等待逻辑
using var semaphore = new Semaphore(0, int.MaxValue, "CrossProcessMsgSemaphore");

// 获取内存映射文件的读写访问器
using var accessor = mmf.CreateViewAccessor();

// 准备要发送的消息
string message = "来自生产者的跨进程消息";
byte[] buffer = Encoding.UTF8.GetBytes(message);

// 锁定内存后写入数据
mutex.WaitOne();
try
{
    accessor.WriteArray(0, buffer, 0, buffer.Length);
}
finally
{
    mutex.ReleaseMutex();
}
// 通知消费端有新数据
semaphore.Release();

消费者示例代码 #

// 打开已存在的命名内存映射文件
using var mmf = MemoryMappedFile.OpenExisting("CrossProcessMsgChannel");
// 打开已存在的命名互斥体
using var mutex = new Mutex(false, "CrossProcessMsgMutex");
// 打开已存在的命名信号量
using var semaphore = Semaphore.OpenExisting("CrossProcessMsgSemaphore");

using var accessor = mmf.CreateViewAccessor();
byte[] buffer = new byte[1024];

// 等待生产者发送信号
semaphore.WaitOne();
// 锁定内存后读取数据
mutex.WaitOne();
try
{
    accessor.ReadArray(0, buffer, 0, buffer.Length);
    string message = Encoding.UTF8.GetString(buffer).Trim('\0');
    Console.WriteLine($"消费端收到消息：{message}");
}
finally
{
    mutex.ReleaseMutex();
}

方案2：封装跨进程通信组件为Channel风格接口 #

如果不想手动实现底层同步逻辑，可以基于System.IO.Pipes（命名管道）这类原生跨进程通信组件，封装出符合.NET Channel接口（IChannelReader<T>、IChannelWriter<T>）的类，让上层代码可以像使用原生Channel一样调用，底层由命名管道负责跨进程数据传输。

简化封装思路示例 #

public class CrossProcessNamedPipeChannel<T> : IChannelReader<T>, IChannelWriter<T>
{
    private readonly NamedPipeServerStream _serverStream;
    private readonly NamedPipeClientStream _clientStream;
    private readonly JsonSerializerOptions _serializerOptions = new();

    // 构造函数根据角色（生产者/消费者）初始化管道流
    public CrossProcessNamedPipeChannel(string pipeName, bool isServer)
    {
        if (isServer)
            _serverStream = new NamedPipeServerStream(pipeName, PipeDirection.InOut);
        else
            _clientStream = new NamedPipeClientStream(".", pipeName, PipeDirection.InOut);
    }

    // 实现IChannelWriter<T>.WriteAsync方法
    public async ValueTask WriteAsync(T item, CancellationToken cancellationToken = default)
    {
        var json = JsonSerializer.Serialize(item, _serializerOptions);
        var buffer = Encoding.UTF8.GetBytes(json);
        if (_serverStream?.IsConnected == true)
            await _serverStream.WriteAsync(buffer, cancellationToken);
        else if (_clientStream?.IsConnected == true)
            await _clientStream.WriteAsync(buffer, cancellationToken);
    }

    // 实现IChannelReader<T>.ReadAsync方法
    public async ValueTask<T> ReadAsync(CancellationToken cancellationToken = default)
    {
        var buffer = new byte[1024];
        int bytesRead = 0;
        if (_serverStream?.IsConnected == true)
            bytesRead = await _serverStream.ReadAsync(buffer, cancellationToken);
        else if (_clientStream?.IsConnected == true)
            bytesRead = await _clientStream.ReadAsync(buffer, cancellationToken);
        
        var json = Encoding.UTF8.GetString(buffer, 0, bytesRead);
        return JsonSerializer.Deserialize<T>(json, _serializerOptions)!;
    }

    // 实现其他接口方法（如Complete、Completion等）
    public void Complete(Exception? error = null)
    {
        _serverStream?.Dispose();
        _clientStream?.Dispose();
    }

    public ValueTask<bool> WaitToReadAsync(CancellationToken cancellationToken = default)
    {
        // 根据管道流的状态实现等待逻辑
        return new ValueTask<bool>(true);
    }

    public Task Completion => Task.CompletedTask;
}

关键注意事项 #

• 跨进程传输对象时，要处理序列化/反序列化问题，推荐使用System.Text.Json，避免使用存在安全风险的BinaryFormatter。
• 命名同步原语或管道的名称要确保唯一性，避免和系统内其他进程的资源冲突。
• 必须妥善处理进程异常退出的情况，用try-finally确保同步原语和管道资源正确释放，防止资源泄漏。

内容的提问来源于stack exchange，提问作者Ricardo Peres