C++操作Windows 管道编程实战

// server.cpp — 最简命名管道服务端
#include <windows.h>
#include <iostream>
#include <string>
#include <io.h>      // _setmode
#include <fcntl.h>   // _o_u16text
const wchar_t* pipe_name = l"\\\\.\\pipe\\mylearningpipe";
int main()
{
    _setmode(_fileno(stdout), _o_u16text);   // 控制台中文设置, 见上文说明
    // 1. 创建管道: 双向 + 消息模式
    handle hpipe = createnamedpipe(
        pipe_name,
        pipe_access_duplex,                          // 双向读写
        pipe_type_message | pipe_readmode_message,   // 消息模式(见第五节)
        1,                                           // 最多 1 个客户端
        4096, 4096, 0, null);
    // 2. 阻塞等待客户端连接
    connectnamedpipe(hpipe, null);
    // 3. 收发循环: 读一条 -> 回一条, 直到收到 "quit"
    wchar_t buf[1024];
    dword n;
    while (readfile(hpipe, buf, sizeof(buf) - sizeof(wchar_t), &n, null) && n > 0)
    {
        buf[n / sizeof(wchar_t)] = l'\0';
        std::wstring msg(buf);
        std::wcout << l"[收到] " << msg << std::endl;
        if (msg == l"quit") break;
        std::wstring reply = l"服务端已收到: " + msg;
        writefile(hpipe, reply.c_str(),
                  (dword)((reply.size() + 1) * sizeof(wchar_t)), &n, null);
    }
    // 4. 收尾
    disconnectnamedpipe(hpipe);
    closehandle(hpipe);
}

// client.cpp — 最简命名管道客户端
#include <windows.h>
#include <iostream>
#include <string>
#include <io.h>      // _setmode
#include <fcntl.h>   // _o_u16text
const wchar_t* pipe_name = l"\\\\.\\pipe\\mylearningpipe";
int main()
{
    _setmode(_fileno(stdout), _o_u16text);   // 控制台中文设置, 见上文说明
    _setmode(_fileno(stdin), _o_u16text);
    // 1. 等服务端的管道可用(最多 5 秒), 否则 createfile 会直接失败
    if (!waitnamedpipe(pipe_name, 5000))
    {
        std::wcout << l"连接失败: 请先启动服务端 (错误码 " << getlasterror() << l")" << std::endl;
        return 1;
    }
    // 2. 打开管道 = 建立连接(和打开文件写法一模一样)
    handle hpipe = createfile(pipe_name, generic_read | generic_write,
                              0, null, open_existing, 0, null);
    // 3. 客户端读模式也要改成消息模式, 和服务端保持一致
    dword mode = pipe_readmode_message;
    setnamedpipehandlestate(hpipe, &mode, null, null);
    // 4. 发一条 -> 收一条回复
    std::wstring line = l"你好, 管道";
    dword n;
    writefile(hpipe, line.c_str(), (dword)((line.size() + 1) * sizeof(wchar_t)), &n, null);
    wchar_t buf[1024];
    readfile(hpipe, buf, sizeof(buf) - sizeof(wchar_t), &n, null);
    buf[n / sizeof(wchar_t)] = l'\0';
    std::wcout << l"[回复] " << buf << std::endl;
    closehandle(hpipe);
}

// multiclientserver.cpp — 循环接受多个客户端(串行版)
#include <windows.h>
#include <iostream>
#include <string>
#include <io.h>      // _setmode
#include <fcntl.h>   // _o_u16text
const wchar_t* pipe_name = l"\\\\.\\pipe\\mylearningpipe";
int main()
{
    _setmode(_fileno(stdout), _o_u16text);   // 控制台中文设置, 见上文说明
    // nmaxinstances 限制的是"同一时刻"最多有多少个客户端连着这个管道。
    // 串行版同一时刻只服务 1 个客户端, 填 1 也够用;
    // 这里先写成 pipe_unlimited_instances, 为后面的并发版做铺垫。
    handle hpipe = createnamedpipe(
        pipe_name,
        pipe_access_duplex,
        pipe_type_message | pipe_readmode_message | pipe_wait,
        pipe_unlimited_instances,
        4096, 4096, 0, null);
    if (hpipe == invalid_handle_value)
        return 1;
    std::wcout << l"服务端启动, 循环等待客户端..." << std::endl;
    for (;;)
    {
        // ---- 接受下一个客户端 ----
        if (!connectnamedpipe(hpipe, null)
            && getlasterror() != error_pipe_connected)
        {
            std::wcerr << l"连接出错: " << getlasterror() << std::endl;
            continue;
        }
        std::wcout << l"客户端已连接!" << std::endl;
        // ---- 服务当前客户端, 直到它断开 ----
        wchar_t buf[1024];
        dword n;
        for (;;)
        {
            bool ok = readfile(hpipe, buf, sizeof(buf) - sizeof(wchar_t), &n, null);
            if (!ok || n == 0)
                break;                          // 客户端断开连接
            buf[n / sizeof(wchar_t)] = l'\0';
            std::wstring msg(buf);
            std::wcout << l"[收到] " << msg << std::endl;
            if (msg == l"quit")
                break;
            std::wstring reply = l"服务端已收到: " + msg;
            writefile(hpipe, reply.c_str(),
                      (dword)((reply.size() + 1) * sizeof(wchar_t)), &n, null);
        }
        // ---- 关键: 断开当前客户端, 准备服务下一个 ----
        // disconnectnamedpipe 不会销毁管道, 之后可以再次 connectnamedpipe
        flushfilebuffers(hpipe);
        disconnectnamedpipe(hpipe);
        std::wcout << l"客户端已断开, 继续等待下一个..." << std::endl;
    }
    closehandle(hpipe);   // 正常跑不到这里, 可自行加退出信号
}

// 并发版骨架: 每接受一个客户端, 就立刻再创建一个实例等下一个
dword winapi clientthread(lpvoid param)
{
    handle hpipe = (handle)param;
    // ... 收发消息, 和串行版内层循环一样 ...
    flushfilebuffers(hpipe);
    disconnectnamedpipe(hpipe);
    closehandle(hpipe);
    return 0;
}
int main()
{
    bool first = true;   // 第一个实例必须带 file_flag_first_pipe_instance
    for (;;)
    {
        handle hpipe = createnamedpipe(
            pipe_name,
            pipe_access_duplex | (first ? file_flag_first_pipe_instance : 0),
            pipe_type_message | pipe_readmode_message | pipe_wait,
            pipe_unlimited_instances,     // 允许多个客户端同时在线
            4096, 4096, 0, null);
        first = false;
        if (hpipe == invalid_handle_value) break;
        if (!connectnamedpipe(hpipe, null)
            && getlasterror() != error_pipe_connected)
        {
            closehandle(hpipe);
            continue;
        }
        handle hthread = createthread(null, 0, clientthread, hpipe, 0, null);
        closehandle(hthread);   // 不等待线程结束, 立刻回去接下一个客户
    }
}

// ============ 消息模式 (pipe_type_message) ============
// 客户端写两条消息:
writefile(hpipe, l"hello", 6 * 2, &n, null);   // 第 1 条消息
writefile(hpipe, l"world", 6 * 2, &n, null);   // 第 2 条消息
// 服务端读两次, 每次恰好读回一条完整消息:
readfile(hpipe, buf1, 1024, &n, null);   // -> "hello"
readfile(hpipe, buf2, 1024, &n, null);   // -> "world"
// ============ 字节模式 (pipe_type_byte) ============
// 同样的两次写, 服务端读一次的结果是"不确定"的:
readfile(hpipe, buf, 1024, &n, null);
// buf 可能是 "helloworld"(全读走了), 也可能是 "hello"(只读到一半),
// 甚至更少 —— 字节流没有消息边界, 一切都看缓冲区的心情

// 父进程: 用匿名管道抓取子进程的输出
#include <windows.h>
#include <iostream>
int main()
{
    handle hread, hwrite;
    security_attributes sa = { sizeof(sa), null, true };  // true = 句柄可被子进程继承
    createpipe(&hread, &hwrite, &sa, 0);                  // 匿名管道: 单向, 字节模式
    startupinfo si = { sizeof(si) };
    si.dwflags = startf_usestdhandles;
    si.hstdoutput = hwrite;   // 子进程往 stdout 写, 实际是写进管道
    si.hstderror  = hwrite;
    process_information pi;
    createprocess(null, (lpwstr)l"child.exe", null, null,
                  true /* 允许继承句柄 */, 0, null, null, &si, &pi);
    closehandle(hwrite);      // 关键! 父进程必须关掉自己这份写端,
                              // 否则子进程退出后, 读端永远等不到"文件结尾"
    char buf[256];
    dword n;
    while (readfile(hread, buf, sizeof(buf) - 1, &n, null) && n > 0)
    {
        buf[n] = 0;
        std::cout << buf;     // 打印子进程的输出
    }
    closehandle(hread);
    waitforsingleobject(pi.hprocess, infinite);
}

 手机 (tcp 客户端)                     pc (接收方)
┌──────────────┐  tcp(9527端口)   ┌─────────────────────────────┐
│ 发送文件数据   │ ───────────────► │ 传输引擎(收文件、算进度)       │
│ 每块报告进度   │                  │    │ 命名管道广播进度          │
└──────────────┘                  │    ▼                        │
                                  │ ui进程1 / ui进程2 / 悬浮窗... │
                                  └─────────────────────────────┘

// 订阅端(ui 进程): 管道名从 \\.\pipe\xxx 换成 \\主机名或ip\pipe\xxx
#include <windows.h>
#include <iostream>
#include <io.h>      // _setmode
#include <fcntl.h>   // _o_u16text
const wchar_t* pipe_name = l"\\\\192.168.1.100\\pipe\\filetransferprogress";
int main()
{
    _setmode(_fileno(stdout), _o_u16text);   // 控制台中文设置, 见第三节说明
    waitnamedpipe(pipe_name, 5000);
    handle hpipe = createfile(pipe_name, generic_read, 0, null, open_existing, 0, null);
    dword mode = pipe_readmode_message;
    setnamedpipehandlestate(hpipe, &mode, null, null);
    wchar_t buf[256];
    dword n;
    while (readfile(hpipe, buf, sizeof(buf) - sizeof(wchar_t), &n, null) && n > 0)
    {
        buf[n / sizeof(wchar_t)] = l'\0';
        std::wcout << l"\r" << buf << std::flush;   // buf = "进度 40%" 之类
    }
    closehandle(hpipe);
}

// progressserver.cpp — 文件传输进度广播服务器
// 场景: 传输引擎一边接收文件数据, 一边把进度广播给所有订阅的 ui 客户端
#include <windows.h>
#include <vector>
#include <string>
#include <algorithm>
const wchar_t* pipe_name = l"\\\\.\\pipe\\filetransferprogress";
critical_section g_cs;                 // 保护客户端列表
std::vector<handle> g_clients;         // 所有已连接客户端的句柄
// 向所有订阅者广播一条消息(消息模式, 一次一条)
void broadcast(const std::wstring& msg)
{
    // 先复制一份句柄列表, 在锁外发送, 避免阻塞 i/o 长时间占着锁
    entercriticalsection(&g_cs);
    std::vector<handle> snapshot = g_clients;
    leavecriticalsection(&g_cs);
    for (handle h : snapshot)
    {
        dword n = 0;
        if (!writefile(h, msg.c_str(), (dword)((msg.size() + 1) * sizeof(wchar_t)), &n, null))
        {
            // 写失败: 客户端已断开, 先从列表移除, 句柄交给 clientthread 清理
            entercriticalsection(&g_cs);
            auto it = std::find(g_clients.begin(), g_clients.end(), h);
            if (it != g_clients.end())
                g_clients.erase(it);
            leavecriticalsection(&g_cs);
        }
    }
}
// 模拟传输引擎的进度产生线程(真实项目里数据来自 tcp 接收循环)
dword winapi progressthread(lpvoid)
{
    for (int pct = 0; pct <= 100; pct += 10)
    {
        broadcast(l"进度 " + std::to_wstring(pct) + l"%");
        sleep(500);
    }
    broadcast(l"传输完成");
    return 0;
}
// 每个客户端一个线程: 用 peeknamedpipe 轮询指令, 顺带检测客户端断开
dword winapi clientthread(lpvoid param)
{
    handle hpipe = (handle)param;
    wchar_t buf[256];
    dword n;
    for (;;)
    {
        // peeknamedpipe 只查看不取走数据, 从不阻塞 —— 这是重点!
        // 如果这里用阻塞的 readfile, 会卡住 broadcast 线程对同一句柄的 writefile
        dword avail = 0;
        bool peeked = peeknamedpipe(hpipe, buf, sizeof(buf), null, &avail, null);
        if (!peeked && getlasterror() != error_insufficient_buffer)
            break;   // 客户端断开(error_insufficient_buffer = 有数据但 peek 缓冲区装不下)
        if (peeked && avail > 0)
        {
            // 有客户端发来的指令, 读出来处理
            if (!readfile(hpipe, buf, sizeof(buf) - sizeof(wchar_t), &n, null) || n == 0)
                break;
            buf[n / sizeof(wchar_t)] = l'\0';
            // 处理 "pause" / "cancel" 等指令
        }
        else
        {
            sleep(100);   // 没有指令, 歇一会儿再查
        }
    }
    // 客户端断开: 从订阅列表移除并清理
    entercriticalsection(&g_cs);
    auto it = std::find(g_clients.begin(), g_clients.end(), hpipe);
    if (it != g_clients.end())
        g_clients.erase(it);
    leavecriticalsection(&g_cs);
    flushfilebuffers(hpipe);
    disconnectnamedpipe(hpipe);
    closehandle(hpipe);
    return 0;
}
int main()
{
    initializecriticalsection(&g_cs);
    // 启动"传输引擎"的进度模拟
    createthread(null, 0, progressthread, null, 0, null);
    // 接受循环: 每连上一个客户端就开一个线程, 同时创建下一个实例等新客户
    bool first = true;
    for (;;)
    {
        handle hpipe = createnamedpipe(
            pipe_name,
            pipe_access_duplex | (first ? file_flag_first_pipe_instance : 0),
            pipe_type_message | pipe_readmode_message | pipe_wait,
            pipe_unlimited_instances,
            4096, 4096, 0, null);
        first = false;
        if (hpipe == invalid_handle_value) break;
        if (!connectnamedpipe(hpipe, null)
            && getlasterror() != error_pipe_connected)
        {
            closehandle(hpipe);
            continue;
        }
        entercriticalsection(&g_cs);
        g_clients.push_back(hpipe);
        leavecriticalsection(&g_cs);
        handle hthread = createthread(null, 0, clientthread, hpipe, 0, null);
        closehandle(hthread);
    }
}

// pc 端: tcp 进度接收服务(手机端用 java/kotlin/swift 开 socket 连过来)
// 协议约定: 每发送完一块文件数据, 发一行 "百分比\n", 如 "45\n"
#include <winsock2.h>
#include <iostream>
#include <cstdlib>   // atoi
#pragma comment(lib, "ws2_32.lib")
int main()
{
    wsadata wsa;
    wsastartup(makeword(2, 2), &wsa);
    socket s = socket(af_inet, sock_stream, ipproto_tcp);
    sockaddr_in addr{};
    addr.sin_family = af_inet;
    addr.sin_addr.s_addr = inaddr_any;      // 监听所有网卡
    addr.sin_port = htons(9527);
    bind(s, (sockaddr*)&addr, sizeof(addr));
    listen(s, 5);
    socket c = accept(s, null, null);       // 手机连上了
    char line[64];
    while (recv(c, line, sizeof(line) - 1, 0) > 0)
    {
        int pct = atoi(line);               // 简单起见, 一行一个百分比
        std::cout << "进度: " << pct << "%\n";
    }
}