C# 多进程管道双向通信

服务端代码

    public class ServerPipe
{
#region 成员变量
/// <summary>
/// 接受名称
/// </summary>
private string _receivePipeName;
/// <summary>
/// 接受数据回调函数
/// </summary>
public event ResultCallBackPoint _resultCallBackEvent;
/// <summary>
/// 服务器端线程
/// </summary>
private Thread pipeServerThread;
/// <summary>
/// 服务器端管道
/// </summary>
private NamedPipeServerStream namedPipeServerStream;
private const int PipeInBufferSize = 4096;
private const int PipeOutBufferSize = 65535;
#endregion
/// <summary>
/// 交互类 结构函数
/// </summary>
/// <param name="sendPort">发送端口名称</param>
/// <param name="receivePort">接受端口名称</param>
public ServerPipe(string receivePipeName)
{
_receivePipeName = receivePipeName;
}
#region 服务器端接受
/// <summary>
/// 接收端
/// </summary>
public void Receive()
{
pipeServerThread = new Thread(PipeServer);
pipeServerThread.IsBackground = true;
pipeServerThread.Start();
}
/// <summary>
/// 服务器
/// </summary>
private void PipeServer()
{
try
{
namedPipeServerStream = new NamedPipeServerStream(_receivePipeName,
PipeDirection.InOut,
1,
PipeTransmissionMode.Message,
PipeOptions.Asynchronous | PipeOptions.WriteThrough,
PipeInBufferSize,
PipeOutBufferSize);
namedPipeServerStream.BeginWaitForConnection(CallBackReceiveData, namedPipeServerStream);
}
catch (Exception ex)
{
CallBackMsg(0, ex.ToString());
}
}
/// <summary>
/// 连接接受回调函数
/// </summary>
/// <param name="result"></param>
private void CallBackReceiveData(IAsyncResult result)
{
try
{
NamedPipeServerStream pipeServer = (NamedPipeServerStream)result.AsyncState;
pipeServer.EndWaitForConnection(result);
var data = new byte[PipeInBufferSize];
var count = pipeServer.Read(data, 0, PipeInBufferSize);
if (count <= 0) return;
// 通信双方可以约定好传输内容的形式，例子中我们传输简单文本信息。
string message = UTF8Encoding.UTF8.GetString(data, 0, count);
CallBackMsg(1, message);
}
catch (Exception ex)
{
CallBackMsg(0, ex.ToString());
}
}
/// <summary>
/// 服务端发送消息
/// </summary>
/// <param name="message"></param>
public void ServerSend(string message)
{
if (!namedPipeServerStream.IsConnected) return;
try
{
byte[] data = UTF8Encoding.UTF8.GetBytes(message);
namedPipeServerStream.Write(data, 0, data.Length);
namedPipeServerStream.Flush();
namedPipeServerStream.WaitForPipeDrain();
}
catch (Exception e)
{
Console.WriteLine(e);
throw;
}
}
public void ResetPipe()
{
namedPipeServerStream.Disconnect();
namedPipeServerStream.BeginWaitForConnection(CallBackReceiveData, namedPipeServerStream);
}
#endregion
#region 委托
/// <summary>
/// 接受数据回调函数委托
/// </summary>
public delegate void ResultCallBackPoint(int code, string msg);
#endregion
/// <summary>
/// 返回信息
/// </summary>
/// <param name="code">成功码</param>
/// <param name="msg">信息</param>
private void CallBackMsg(int code, string msg)
{
if (_resultCallBackEvent != null)
{
_resultCallBackEvent(code, msg);
}
}
}

客户端代码:

    /// <summary>
/// 通道客户端
/// </summary>
public class ClientPipe
{
#region 成员变量
/// <summary>
/// 发送消息
/// </summary>
private string _sendData;
/// <summary>
/// 接受数据回调函数
/// </summary>
public event ResultCallBackPoint resultClientCallBackEvent;
/// <summary>
/// 客户端线程
/// </summary>
private Thread pipeCliendThread;
/// <summary>
/// 客户端管道
/// </summary>
private NamedPipeClientStream namedPipeClientStream;
private string _sendPipeName;
#endregion
#region 委托
/// <summary>
/// 接受数据回调函数委托
/// </summary>
public delegate void ResultCallBackPoint(int code, string msg);
#endregion
public ClientPipe(string sendPipeName)
{
this._sendPipeName = sendPipeName;
}
#region 客户端发送
/// <summary>
/// 客户端通道
/// </summary>
private void PipeClient()
{
try
{
namedPipeClientStream = new NamedPipeClientStream(".", _sendPipeName, PipeDirection.InOut, PipeOptions.Asynchronous | PipeOptions.WriteThrough);
namedPipeClientStream.Connect(10000);
namedPipeClientStream.ReadMode = PipeTransmissionMode.Message;
byte[] data = UTF8Encoding.UTF8.GetBytes(_sendData);
//namedPipeClientStream.BeginWrite(data, 0, data.Length, ClientPipeWriteCallback, namedPipeClientStream);
namedPipeClientStream.BeginWrite(data, 0, data.Length, endWriteIAsyncResult =>
{
try
{
// Get the pipe
NamedPipeClientStream endWriteCallbackNamedPipeClientStream = (NamedPipeClientStream)endWriteIAsyncResult.AsyncState;
// End the write
endWriteCallbackNamedPipeClientStream?.EndWrite(endWriteIAsyncResult);
endWriteCallbackNamedPipeClientStream?.Flush();
byte[] readBuffer = new byte[1024];
endWriteCallbackNamedPipeClientStream?.BeginRead(readBuffer, 0, readBuffer.Length, endReadIAsyncResult =>
{
try
{
// Get the pipe
NamedPipeClientStream endReadCallbackNamedPipeClientStream = (NamedPipeClientStream)endReadIAsyncResult.AsyncState;
// End the Read
var readBytes = endReadCallbackNamedPipeClientStream?.EndRead(endReadIAsyncResult);
string message = Encoding.UTF8.GetString(readBuffer, 0, (int)readBytes);
//if (endReadCallbackNamedPipeClientStream != null)
//{
//}
if (!string.IsNullOrWhiteSpace(message))
{
ClientCallBackMsg(1, message);
}
}
catch (Exception catchException)
{
throw catchException;
}
finally
{
}
}, endWriteCallbackNamedPipeClientStream);
}
catch (Exception catchException)
{
throw catchException;
}
finally
{
}
}, namedPipeClientStream);
}
catch (Exception ex)
{
ClientCallBackMsg(0, ex.ToString());
}
}
/// <summary>
/// 发送
/// </summary>
public void Send(string sendData)
{
_sendData = sendData;
pipeCliendThread = new Thread(PipeClient);
pipeCliendThread.IsBackground = true;
pipeCliendThread.Start();
}
#endregion
/// <summary>
/// 客户端返回信息
/// </summary>
/// <param name="code">成功码</param>
/// <param name="msg">信息</param>
private void ClientCallBackMsg(int code, string msg)
{
if (resultClientCallBackEvent != null)
{
resultClientCallBackEvent(code, msg);
}
pipeCliendThread = null;
}
}

服务端使用代码:

        ServerPipe server = null;
public MainWindow()
{
InitializeComponent();
server = new ServerPipe("test1");
server._resultCallBackEvent += (code, msg) =>
{
Console.WriteLine($"{code}-----{msg}");
if (msg.IndexOf("init")>-1)
{
server.ServerSend("How are you" + DateTime.Now.ToString("O"));
}
server.ResetPipe();
};
server.Receive();
}

客户端使用代码

ClientPipe client = new ClientPipe("test1");
client.resultClientCallBackEvent += (code, msg) =>
{
Console.WriteLine($"{code}-----{msg}");
};
System.Timers.Timer timer1 = null;
timer1 = new System.Timers.Timer();
timer1.Elapsed += delegate
{
client.Send("init" + DateTime.Now.ToString("O"));
};
timer1.Interval = 1000; // 每1秒重复检测一次
timer1.Start();

结果：