如何正确实现Node.js Unix域套接字通信？排查高负载乱码问题

问题背景 #

我正在调试多线程Node.js实例间的IPC实现，由于Node.js原生不支持数据报套接字，采用默认流协议+应用层封装的方式，服务器端持续接收、客户端持续发送。

现有代码实现 #

客户端发送逻辑

// const transmitter = net.createConnection(SOCKETFILE);

const outgoing_buffer = [];
let writeable = true;

const write = (transfer) => {
    if (transfer) outgoing_buffer.push(transfer);
    if (outgoing_buffer.length === 0) return;
    if (!writeable) return;
    const current = outgoing_buffer.shift();
    writeable = false;
    transmitter.write(current, "utf8", () => {
        writeable = true;
        write();
    });
};

服务器端接收逻辑

// const server = net.createServer();
// server.listen(SOCKETFILE);
// server.on("connection", (reciever) => { ...
// reciever.on("data", (data) => { ...
// ... 调用read函数处理数据

let incoming_buffer = "";
const read = (data) => {
    incoming_buffer += data.toString();
    while (true) {
        const decoded = decode(incoming_buffer);
        if (!decoded) return;
        incoming_buffer = incoming_buffer.substring(decoded.length);
        // ... 处理解码后的字符串
    }
};

问题现象 #

数据流通过JSON序列化后传输，但偶尔会在incoming_buffer中出现随机乱码（显示为���），且高CPU负载时该现象更频繁。即使概率极低，也严重影响通信可靠性，需要一种在CPU满负载时仍能保证数据流无损坏、无意外字符的可靠实现。

潜在成因 #

• UTF-8字节截断：Node.js net模块的data事件传递的是Buffer，高负载下TCP流会拆分包，若直接将Buffer转成字符串拼接，多字节UTF-8字符（如中文、特殊符号）可能被拆分在两个data事件中，导致截断的字节被解析为无效字符（�）。
• 应用层协议无明确边界：现有decode函数依赖字符串内容拆分数据包，若数据包不完整（仅收到部分字节），字符串解析时会破坏UTF-8结构，产生乱码。
• 字符串缓冲区的不可靠性：用字符串拼接缓冲区，本质上是多次将Buffer转码为字符串，过程中若遇到不完整的UTF-8序列，Node.js会直接替换为�，且无法恢复原始字节。

可靠实现方案 #

1. 改用字节缓冲区维护原始数据 #

放弃字符串缓冲区，直接用Buffer存储未处理的字节，确保UTF-8字符的完整性：

let incoming_buffer = Buffer.alloc(0);
const read = (data) => {
    // 拼接原始字节缓冲区
    incoming_buffer = Buffer.concat([incoming_buffer, data]);
    
    while (true) {
        // 假设decode函数接收Buffer，返回{ data: 解码结果, consumed: 消耗字节数 }，无完整包时返回null
        const decoded = decode(incoming_buffer);
        if (!decoded) break;
        
        // 裁剪已处理的字节
        incoming_buffer = incoming_buffer.slice(decoded.consumed);
        // 处理解码后的数据
        // ...
    }
};

2. 定义带长度前缀的应用层协议 #

通过固定长度的前缀（如4字节无符号整数）明确每个数据包的字节长度，从根本上避免分包导致的解析错误：

客户端发送逻辑优化

const write = (transfer) => {
    if (transfer) outgoing_buffer.push(transfer);
    if (outgoing_buffer.length === 0 || !writeable) return;
    
    const current = outgoing_buffer.shift();
    const jsonBuffer = Buffer.from(JSON.stringify(current), 'utf8');
    // 生成4字节大端序的长度前缀
    const lengthPrefix = Buffer.alloc(4);
    lengthPrefix.writeUInt32BE(jsonBuffer.length, 0);
    
    writeable = false;
    // 合并前缀与数据，一次性写入
    transmitter.write(Buffer.concat([lengthPrefix, jsonBuffer]), () => {
        writeable = true;
        write();
    });
};

服务器端接收逻辑优化

let incoming_buffer = Buffer.alloc(0);
const read = (data) => {
    incoming_buffer = Buffer.concat([incoming_buffer, data]);
    
    // 先检查是否有完整的长度前缀（4字节）
    while (incoming_buffer.length >= 4) {
        const dataLength = incoming_buffer.readUInt32BE(0);
        // 检查是否有足够的数据包字节
        if (incoming_buffer.length < 4 + dataLength) {
            break; // 数据不完整，等待下一个data事件
        }
        
        // 提取完整的数据包字节并解析
        const jsonBuffer = incoming_buffer.slice(4, 4 + dataLength);
        try {
            const decodedData = JSON.parse(jsonBuffer.toString('utf8'));
            // 处理解码后的数据
            // ...
        } catch (err) {
            console.error('数据包解析失败:', err);
            // 错误处理：可重置缓冲区或跳过错误包
            incoming_buffer = incoming_buffer.slice(4 + dataLength);
            continue;
        }
        
        // 裁剪已处理的字节
        incoming_buffer = incoming_buffer.slice(4 + dataLength);
    }
};

3. 增强错误处理 #

添加JSON解析的异常捕获，避免单个损坏数据包导致整个接收逻辑崩溃，同时可根据需求实现错误恢复逻辑。

内容的提问来源于stack exchange，提问作者Király István