基于Named Pipes的IPC程序中服务端与客户端通信故障排查

我用C#开发了两个独立解决方案的Windows Forms应用，目标是基于命名管道实现进程间通信（IPC）的类消息应用，作为IPC技术的练习项目。但目前服务端与客户端无法互发消息，程序无法正常运行，不确定问题出在命名管道配置还是线程管理上。

服务端代码 #

public partial class serverForm : Form
{
    private NamedPipeServerStream pipeServer = null;
    private StreamReader streamReader = null;
    private StreamWriter streamWriter = null;
    private Semaphore semaphore = new Semaphore(initialCount: 1, maximumCount: 1);
    private bool startWork = false;
    private String sendMsg = null;

    public serverForm()
    {
        InitializeComponent();
    }

    private void serverForm_Load(object sender, EventArgs e)
    {
       new Thread(() =>
        {
            StartServer();
        }).Start();
    }

    private void StartServer()
    {
        pipeServer = new NamedPipeServerStream("testpipe", PipeDirection.InOut);

        pipeServer.WaitForConnection();

        startWork = true;

        streamReader = new StreamReader(pipeServer);
        streamWriter = new StreamWriter(pipeServer);

        Thread thread2 = new Thread(() => StreamReader());
        thread2.Start();

        while (pipeServer.IsConnected) 
        {
            if (semaphore.WaitOne())
            {
                if (sendMsg != null && sendMsg.Length > 0)
                {
                    streamWriter.WriteLine(sendMsg);
                    streamWriter.Flush();
                    sendMsg = null;
                }
                semaphore.Release();
            }                

            Thread.Sleep(500);
        }
    }

    private void StreamWrite(String msg)
    {
        //StreamWriter writer = new StreamWriter(pipeServer);
        //writer.Write(msg);
        //writer.Close();

        //streamWriter.WriteLine(msg);
        //streamWriter.Flush();

        while (pipeServer.IsConnected)
        {
            if (semaphore.WaitOne())
            {
                sendMsg = msg;
                semaphore.Release();

                break;
            }

            Thread.Sleep(50);
        }
    }

    private void StreamReader()
    {
        while (pipeServer != null && pipeServer.IsConnected)
        {
            string data = streamReader.ReadLine();

            if (data != null)
            {
                messagesTxt.BeginInvoke((Action)(() =>
                {
                    messagesTxt.Text += "Client: " + data + "\r\n";
                }));
            }
        }
    }

    private void sendMessageTxt_TextChanged(object sender, EventArgs e)
    {

    }

    private void sendMessageBtn_Click(object sender, EventArgs e)
    {
        if (startWork)
        {
            String msg = sendMessageTxt.Text;
            if (msg != null && msg.Length > 0)
            {
                messagesTxt.Text += "Server: " + msg + "\r\n";
                Thread thread1 = new Thread(() => StreamWrite(msg));
                thread1.Start();
                
                sendMessageTxt.Text = "";
            }
        }
        else
        {
            MessageBox.Show(this, "Connection is not valid");
        }
    }

    private void sendMessageTxt_KeyUp(object sender, KeyEventArgs e)
    {
        if (e.KeyCode == Keys.Return)
        {
            sendMessageBtn_Click(sender, null);
        }
    }
}

客户端代码 #

public partial class clientForm : Form
{
    private NamedPipeClientStream pipeClient = null;
    private StreamReader streamReader = null;
    private StreamWriter streamWriter = null;
    private Semaphore semaphore = new Semaphore(initialCount: 1, maximumCount: 1);
    private bool startWork = false;
    private String sendMsg = null;
    private string serverMessage;
    private string defaultClientMessage = "No Message Yet";
    public clientForm()
    {
        InitializeComponent();
    }

    private void clientForm_Load(object sender, EventArgs e)
    {
        new Thread(() =>
        {
            StartServer();
        }).Start();
    }

    private void StartServer()
    {
        pipeClient = new NamedPipeClientStream(".", "testpipe", PipeDirection.InOut);

        pipeClient.Connect();

        startWork = true;


        streamReader = new StreamReader(pipeClient);
        streamWriter = new StreamWriter(pipeClient);

        Thread thread2 = new Thread(() => StreamReader(pipeClient));
        thread2.Start();

        while (pipeClient.IsConnected)
        {
            if (semaphore.WaitOne())
            {
                if (sendMsg != null && sendMsg.Length > 0)
                {
                    streamWriter.WriteLine(sendMsg);
                    streamWriter.Flush();
                    sendMsg = null;
                }
                semaphore.Release();
            }

            Thread.Sleep(500);
        }
    }

    private void StreamReader(NamedPipeClientStream pipeClient)
    {
        while (pipeClient != null && pipeClient.IsConnected)
        {
            string data = streamReader.ReadLine();

            if (data != null)
            {
                messagesTxt.BeginInvoke((Action)(() =>
                {
                    messagesTxt.Text += "Client: " + data + "\r\n";
                }));
            }
        }
    }

    private void StreamWrite(NamedPipeClientStream pipeClient, String msg)
    {
        while (pipeClient.IsConnected)
        {
            if (semaphore.WaitOne())
            {
                sendMsg = msg;
                semaphore.Release();

                break;
            }

            Thread.Sleep(50);
        }
    }

    private void sendMessageBtn_Click(object sender, EventArgs e)
    {
        if (startWork)
        {
            String msg = sendMessageTxt.Text;
            if (msg != null && msg.Length > 0)
            {
                messagesTxt.Text += "Client: " + msg + "\r\n";
                Thread thread1 = new Thread(() => StreamWrite(pipeClient, msg));
                thread1.Start();
                

                sendMessageTxt.Text = "";
            }
        }
        else
        {
            MessageBox.Show(this, "Connection is not valid");
        }
    }

    private void sendMessageTxt_KeyUp(object sender, KeyEventArgs e)
    {
        if (e.KeyCode == Keys.Return)
        {
            sendMessageBtn_Click(sender, null);
        }
    }
}

问题排查与修复方案 #

核心问题点 #

1. 客户端消息显示错误：客户端StreamReader方法中，将收到的服务端消息前缀写为"Client: "，导致无法区分消息来源，应改为"Server: "。
2. 流写入未自动刷新：创建StreamWriter时未开启自动刷新，虽然手动调用了Flush，但仍可能导致消息滞留，需初始化时设置AutoFlush = true。
3. 发送逻辑冗余：通过中间变量sendMsg加500ms轮询的发送逻辑完全多余，直接在发送时通过信号量保护流写入即可。
4. 连接状态判断不准确：IsConnected属性无法实时反映连接状态，需通过读写操作的异常来判断连接是否有效。
5. 资源未释放：窗体关闭时未释放管道、流等资源，易造成资源泄漏。

关键代码修改 #

服务端修改

• 初始化StreamWriter时开启自动刷新：

streamWriter = new StreamWriter(pipeServer) { AutoFlush = true };

• 简化发送逻辑，去掉轮询和中间变量：

private void StreamWrite(String msg)
{
    if (!pipeServer.IsConnected) return;
    if (semaphore.WaitOne())
    {
        try
        {
            streamWriter.WriteLine(msg);
        }
        finally
        {
            semaphore.Release();
        }
    }
}

• 发送按钮点击时无需新开线程，直接调用发送方法：

private void sendMessageBtn_Click(object sender, EventArgs e)
{
    if (startWork)
    {
        String msg = sendMessageTxt.Text;
        if (!string.IsNullOrEmpty(msg))
        {
            messagesTxt.Text += "Server: " + msg + "\r\n";
            StreamWrite(msg);
            sendMessageTxt.Text = "";
        }
    }
    else
    {
        MessageBox.Show(this, "连接无效");
    }
}

客户端修改

• 修正消息显示前缀：

messagesTxt.BeginInvoke((Action)(() =>
{
    messagesTxt.Text += "Server: " + data + "\r\n";
}));

• 同样设置StreamWriter自动刷新，并简化发送逻辑，添加窗体关闭时的资源释放：

protected override void OnFormClosing(FormClosingEventArgs e)
{
    base.OnFormClosing(e);
    streamReader?.Close();
    streamWriter?.Close();
    pipeClient?.Close();
}

额外注意事项 #

• 必须先启动服务端，再启动客户端，确保服务端先进入等待连接状态。
• 命名管道名称需保持一致，本地通信使用"."作为服务器名是正确的。
• 线程操作UI控件必须使用BeginInvoke，这部分原有代码逻辑正确，需保留。

内容的提问来源于stack exchange，提问作者alireza haajalii