Node.js Modbus开发实战完整指南：modbus-serial库 TCP/RTU客户端 数据解析与MQTT上云网关

Node.js以其异步非阻塞、事件驱动的特性，非常适合开发工业数据采集和Modbus通信网关。modbus-serial是Node.js生态中最流行、最活跃的Modbus库，纯JavaScript实现，支持Modbus RTU（串口）、TCP和RTU-over-TCP，同时提供客户端和服务端API。本文从安装配置、TCP/RTU客户端、数据类型解析、服务端模拟、批量采集、MQTT上云到工业级网关实战，全面讲解Node.js Modbus开发。

一、modbus-serial库概述

1.1 什么是modbus-serial

modbus-serial是一个纯JavaScript实现的Modbus协议库，由Yaacov Zamir维护，在npm上每周下载量超过10万次，是Node.js生态中最成熟的Modbus解决方案。它同时支持Modbus RTU（通过串口）、Modbus TCP和RTU-over-TCP三种传输方式，提供Promise和回调两种API风格，并内置了Modbus服务端（从站）模拟功能。

1.2 核心特性

• 多协议支持：Modbus RTU（串口）、Modbus TCP、RTU-over-TCP
• 客户端+服务端：既可以做主站采集数据，也可以做从站模拟设备
• Promise API：所有操作返回Promise，支持async/await
• 数据类型解析：内置32位整数、浮点数、字符串等多寄存器数据解析
• 串口支持：基于node-serialport，支持Windows/Linux/macOS
• 连接管理：自动重连、超时控制、事务ID管理
• 纯JavaScript：无需编译原生模块（TCP模式），跨平台
• 活跃维护：持续更新，issue响应快，社区活跃

1.3 支持的功能码

二、安装与快速开始

# 安装modbus-serial
npm install modbus-serial

# 如果需要RTU串口支持，还需安装serialport
npm install serialport

# 验证安装
node -e "const ModbusRTU = require('modbus-serial'); console.log('modbus-serial已安装');"

// quick_start.js - 5分钟快速上手
const ModbusRTU = require("modbus-serial");

// 创建客户端实例
const client = new ModbusRTU();

// 设置从站地址（单元ID）
client.setID(1);

// 设置超时（毫秒）
client.setTimeout(5000);

// 连接Modbus TCP设备
async function main() {
    try {
        // 连接到Modbus TCP从站
        await client.connectTCP("192.168.1.100", { port: 502 });
        console.log("已连接到Modbus TCP设备");

        // 读取10个保持寄存器（从地址0开始）
        const result = await client.readHoldingRegisters(0, 10);
        console.log("保持寄存器:", result.data);

        // 写入单个寄存器（地址100，值1234）
        await client.writeRegister(100, 1234);
        console.log("已写入寄存器100=1234");

        // 读取8个线圈
        const coils = await client.readCoils(0, 8);
        console.log("线圈状态:", coils.data);

        // 关闭连接
        client.close();
        console.log("连接已关闭");
    } catch (err) {
        console.error("操作失败:", err.message);
    }
}

main();

三、Modbus TCP客户端详解

// tcp_client.js - Modbus TCP客户端完整示例
const ModbusRTU = require("modbus-serial");

class ModbusTcpClient {
    constructor(host, port = 502, unitId = 1, timeout = 5000) {
        this.client = new ModbusRTU();
        this.host = host;
        this.port = port;
        this.unitId = unitId;
        this.timeout = timeout;
        this.connected = false;
    }

    // 连接
    async connect() {
        this.client.setID(this.unitId);
        this.client.setTimeout(this.timeout);

        await this.client.connectTCP(this.host, {
            port: this.port,
            // 可选：自动重连
            // autoReconnect: true,
            // reconnectInterval: 1000,
        });
        this.connected = true;
        console.log(`已连接 ${this.host}:${this.port}, 从站ID=${this.unitId}`);
    }

    // 读保持寄存器
    async readHolding(address, count) {
        const result = await this.client.readHoldingRegisters(address, count);
        return result.data;  // 返回number数组
    }

    // 读输入寄存器
    async readInput(address, count) {
        const result = await this.client.readInputRegisters(address, count);
        return result.data;
    }

    // 读线圈
    async readCoils(address, count) {
        const result = await this.client.readCoils(address, count);
        return result.data;  // 返回boolean数组
    }

    // 读离散输入
    async readDiscrete(address, count) {
        const result = await this.client.readDiscreteInputs(address, count);
        return result.data;
    }

    // 写单个寄存器
    async writeRegister(address, value) {
        await this.client.writeRegister(address, value);
    }

    // 写多个寄存器
    async writeRegisters(address, values) {
        await this.client.writeRegisters(address, values);
    }

    // 写单个线圈
    async writeCoil(address, value) {
        await this.client.writeCoil(address, value);  // value: true/false
    }

    // 关闭
    close() {
        this.client.close();
        this.connected = false;
    }
}

// 使用示例
async function main() {
    const mb = new ModbusTcpClient("192.168.1.100", 502, 1);
    await mb.connect();

    // 读取传感器数据
    const temp = await mb.readHolding(0, 4);
    console.log("温度:", temp[0] / 10, "°C");
    console.log("湿度:", temp[1] / 10, "%");
    console.log("电压:", temp[2] / 10, "V");
    console.log("电流:", temp[3] / 100, "A");

    // 控制设备
    await mb.writeCoil(0, true);   // 启动电机
    await mb.writeRegister(10, 50); // 设置频率50Hz

    mb.close();
}

main().catch(console.error);

四、Modbus RTU串口客户端

// rtu_client.js - Modbus RTU串口客户端
const ModbusRTU = require("modbus-serial");

async function main() {
    const client = new ModbusRTU();
    client.setID(1);        // 从站地址
    client.setTimeout(5000); // 超时5秒

    // 连接串口（Linux: /dev/ttyUSB0, Windows: COM3, macOS: /dev/tty.usbserial-xxx）
    await client.connectRTU("/dev/ttyUSB0", {
        baudRate: 9600,      // 波特率
        dataBits: 8,         // 数据位
        parity: "none",      // 校验位: none/even/odd
        stopBits: 1,         // 停止位
        // 可选：RS485自动收发控制
        // rtsControl: "hardware",
    });
    console.log("RTU串口已连接");

    // 读取保持寄存器
    const result = await client.readHoldingRegisters(0, 10);
    console.log("寄存器数据:", result.data);

    // 读取多个从站（总线上多个设备）
    for (let id = 1; id <= 5; id++) {
        client.setID(id);
        try {
            const data = await client.readHoldingRegisters(0, 4);
            console.log(`从站${id}:`, data.data);
        } catch (err) {
            console.log(`从站${id}无响应:`, err.message);
        }
    }

    client.close();
}

main().catch(console.error);

4.1 串口设备查找

// list_ports.js - 查找可用串口
const SerialPort = require("serialport");

async function listSerialPorts() {
    const ports = await SerialPort.list();
    console.log("可用串口:");
    ports.forEach(port => {
        console.log(`  路径: ${port.path}`);
        console.log(`  制造商: ${port.manufacturer || "未知"}`);
        console.log(`  序列号: ${port.serialNumber || "无"}`);
        console.log(`  pnpId: ${port.pnpId || "无"}`);
        console.log("---");
    });
}

listSerialPorts();

// 常见串口路径：
// Linux:   /dev/ttyUSB0 (USB转RS485), /dev/ttyS0 (板载串口)
// Windows: COM3, COM4
// macOS:   /dev/tty.usbserial-A123456, /dev/cu.usbserial-A123456

五、数据类型解析

Modbus寄存器是16位的，但实际设备经常用多个寄存器组合存储32位整数、浮点数、字符串等数据。modbus-serial提供了buffer工具类来解析这些数据。

// data_types.js - 多寄存器数据类型解析
const ModbusRTU = require("modbus-serial");

// 方法一：使用modbus-serial内置的buffer工具
async function readDataTypes(client) {
    // 读取8个寄存器（足够存放各种数据类型）
    const result = await client.readHoldingRegisters(0, 8);
    const regs = result.data;  // [r0, r1, r2, r3, r4, r5, r6, r7]

    // ====== 16位整数（单个寄存器） ======
    const uint16 = regs[0];           // 无符号16位
    const int16 = regs[0] > 32767 ? regs[0] - 65536 : regs[0];  // 有符号16位

    // ====== 32位整数（两个寄存器） ======
    // 大端模式（高字在前）：value = (r0 << 16) | r1
    const uint32_be = (regs[0] << 16) | regs[1];
    // 小端模式（低字在前）：value = (r1 << 16) | r0
    const uint32_le = (regs[1] << 16) | regs[0];

    // ====== 32位浮点数（两个寄存器，IEEE 754） ======
    // 使用Buffer转换
    const buf = Buffer.alloc(4);
    buf.writeUInt16BE(regs[2], 0);  // 高字
    buf.writeUInt16BE(regs[3], 2);  // 低字
    const float32 = buf.readFloatBE(0);  // 大端浮点数

    // ====== 64位浮点数（四个寄存器） ======
    const buf64 = Buffer.alloc(8);
    for (let i = 0; i < 4; i++) {
        buf64.writeUInt16BE(regs[4 + i], i * 2);
    }
    const double64 = buf64.readDoubleBE(0);

    // ====== 字符串（多个寄存器，每个寄存器存2个ASCII字符） ======
    let str = "";
    for (let i = 0; i < 4; i++) {
        str += String.fromCharCode((regs[i] >> 8) & 0xFF);
        str += String.fromCharCode(regs[i] & 0xFF);
    }
    str = str.replace(/\0/g, "").trim();  // 去掉空字符

    console.log("16位无符号:", uint16);
    console.log("16位有符号:", int16);
    console.log("32位无符号(大端):", uint32_be);
    console.log("32位浮点数:", float32);
    console.log("64位浮点数:", double64);
    console.log("字符串:", str);
}

// 方法二：使用modbus-serial的readHoldingRegisters返回的buffer
// result.buffer 是一个Buffer对象，可以直接用标准方法读取
async function readWithBuffer(client) {
    const result = await client.readHoldingRegisters(0, 8);
    const buf = result.buffer;  // Buffer对象

    const uint16 = buf.readUInt16BE(0);
    const uint32 = buf.readUInt32BE(2);
    const float = buf.readFloatBE(6);

    console.log("使用Buffer读取:", uint16, uint32, float);
}

六、Modbus服务端（从站）模拟

modbus-serial不仅可以做客户端，还可以创建Modbus TCP服务端（从站），用于模拟设备或搭建数据网关。

// modbus_server.js - Modbus TCP服务端（从站）模拟
const ModbusRTU = require("modbus-serial");

// 创建服务端
const server = new ModbusRTU.Server();

// ====== 定义数据向量（回调函数） ======
// 当主站读取时，这些回调函数被调用，返回对应数据
const vector = {
    // 读保持寄存器（功能码03）
    getHoldingRegister: function(addr, unitID, callback) {
        // addr: 寄存器地址
        // unitID: 从站地址
        // callback(err, value): 返回寄存器值
        const holdingRegs = [250, 650, 2200, 135, 0, 0, 0, 0];
        if (addr >= 0 && addr < holdingRegs.length) {
            callback(null, holdingRegs[addr]);
        } else {
            callback({
                exceptionCode: 2,  // 非法地址
            });
        }
    },

    // 读输入寄存器（功能码04）
    getInputRegister: function(addr, unitID, callback) {
        const inputRegs = [100, 200, 300, 400];
        if (addr >= 0 && addr < inputRegs.length) {
            callback(null, inputRegs[addr]);
        } else {
            callback({ exceptionCode: 2 });
        }
    },

    // 读线圈（功能码01）
    getCoil: function(addr, unitID, callback) {
        const coils = [true, false, true, true, false, false, false, false];
        if (addr >= 0 && addr < coils.length) {
            callback(null, coils[addr]);
        } else {
            callback({ exceptionCode: 2 });
        }
    },

    // 读离散输入（功能码02）
    getDiscreteInput: function(addr, unitID, callback) {
        callback(null, addr % 2 === 0);
    },

    // 写单个寄存器（功能码06）
    setRegister: function(addr, value, unitID, callback) {
        console.log(`写寄存器: 地址=${addr}, 值=${value}, 从站=${unitID}`);
        // 在这里执行实际动作，如控制设备
        callback(null);  // 写入成功
    },

    // 写多个寄存器（功能码16）
    setRegisterArray: function(addr, values, unitID, callback) {
        console.log(`批量写寄存器: 地址=${addr}, 值=${values}`);
        callback(null);
    },

    // 写单个线圈（功能码05）
    setCoil: function(addr, value, unitID, callback) {
        console.log(`写线圈: 地址=${addr}, 值=${value}`);
        callback(null);
    },
};

// 启动TCP服务端，监听502端口
server.listenTCP(502, vector, {
    host: "0.0.0.0",
    // 可选：允许的从站ID
    // unitID: 1,
});

console.log("Modbus TCP服务端已启动，监听端口502");
console.log("保持寄存器0-3: 温度25.0°C, 湿度65.0%, 电压220.0V, 电流1.35A");
console.log("按 Ctrl+C 停止");

// 模拟传感器数据实时更新
setInterval(() => {
    // 可以在这里更新内部数据
    // 例如从真实传感器读取数据
}, 1000);

七、批量数据采集与轮询

// poller.js - 工业级批量数据采集器
const ModbusRTU = require("modbus-serial");

class ModbusPoller {
    constructor(config) {
        this.client = new ModbusRTU();
        this.config = config;
        this.data = {};        // 采集到的数据
        this.running = false;
        this.timer = null;
    }

    // 连接
    async connect() {
        this.client.setID(this.config.unitId || 1);
        this.client.setTimeout(this.config.timeout || 5000);

        if (this.config.type === "tcp") {
            await this.client.connectTCP(this.config.host, {
                port: this.config.port || 502,
            });
        } else if (this.config.type === "rtu") {
            await this.client.connectRTU(this.config.port, {
                baudRate: this.config.baudRate || 9600,
                parity: this.config.parity || "none",
            });
        }
        console.log("已连接到Modbus设备");
    }

    // 采集单个寄存器组
    async readGroup(group) {
        try {
            const result = await this.client.readHoldingRegisters(
                group.address, group.count
            );
            // 按配置解析数据
            group.points.forEach((point, idx) => {
                let value = result.data[idx];
                // 应用缩放系数
                if (point.scale) value = value * point.scale;
                // 应用偏移
                if (point.offset) value = value + point.offset;
                // 保留小数位
                if (point.decimals !== undefined) {
                    value = Number(value.toFixed(point.decimals));
                }
                this.data[point.name] = value;
            });
            return true;
        } catch (err) {
            console.error(`采集 ${group.name} 失败:`, err.message);
            return false;
        }
    }

    // 启动周期性采集
    start(intervalMs = 1000) {
        this.running = true;
        const poll = async () => {
            if (!this.running) return;
            for (const group of this.config.groups) {
                await this.readGroup(group);
            }
            // 输出当前数据
            console.log(JSON.stringify(this.data, null, 2));
            this.timer = setTimeout(poll, intervalMs);
        };
        poll();
    }

    // 停止
    stop() {
        this.running = false;
        if (this.timer) clearTimeout(this.timer);
        this.client.close();
    }
}

// 配置示例：采集一个智能电表
const config = {
    type: "tcp",
    host: "192.168.1.100",
    port: 502,
    unitId: 1,
    groups: [
        {
            name: "电参数",
            address: 0,
            count: 6,
            points: [
                { name: "电压", scale: 0.1, decimals: 1, unit: "V" },
                { name: "电流", scale: 0.01, decimals: 2, unit: "A" },
                { name: "有功功率", scale: 0.001, decimals: 3, unit: "kW" },
                { name: "无功功率", scale: 0.001, decimals: 3, unit: "kvar" },
                { name: "功率因数", scale: 0.001, decimals: 3 },
                { name: "频率", scale: 0.01, decimals: 2, unit: "Hz" },
            ],
        },
        {
            name: "电能",
            address: 100,
            count: 2,
            points: [
                { name: "正向有功电能", scale: 0.01, decimals: 2, unit: "kWh" },
                { name: "反向有功电能", scale: 0.01, decimals: 2, unit: "kWh" },
            ],
        },
    ],
};

// 运行
async function main() {
    const poller = new ModbusPoller(config);
    await poller.connect();
    poller.start(2000);  // 每2秒采集一次
}

main().catch(console.error);

八、Modbus转MQTT上云实战

// modbus_mqtt_gateway.js - Modbus转MQTT物联网网关
const ModbusRTU = require("modbus-serial");
const mqtt = require("mqtt");

// ====== 配置 ======
const MODBUS_CONFIG = {
    host: "192.168.1.100",
    port: 502,
    unitId: 1,
};

const MQTT_CONFIG = {
    url: "mqtt://broker.emqx.io",
    port: 1883,
    topic: "modbus/gateway/device001",
    clientId: "modbus_gateway_" + Math.random().toString(16).substr(2, 8),
};

const POLL_INTERVAL = 5000;  // 5秒采集一次

// ====== 数据点配置 ======
const DATA_POINTS = [
    { name: "temperature", address: 0, scale: 0.1, unit: "°C" },
    { name: "humidity", address: 1, scale: 0.1, unit: "%" },
    { name: "voltage", address: 2, scale: 0.1, unit: "V" },
    { name: "current", address: 3, scale: 0.01, unit: "A" },
    { name: "power", address: 4, scale: 0.001, unit: "kW" },
    { name: "energy", address: 100, scale: 0.01, unit: "kWh" },
];

// ====== 主程序 ======
async function main() {
    // 1. 连接Modbus
    const mbClient = new ModbusRTU();
    mbClient.setID(MODBUS_CONFIG.unitId);
    mbClient.setTimeout(3000);
    await mbClient.connectTCP(MODBUS_CONFIG.host, { port: MODBUS_CONFIG.port });
    console.log("Modbus已连接");

    // 2. 连接MQTT
    const mqttClient = mqtt.connect(MQTT_CONFIG.url, {
        clientId: MQTT_CONFIG.clientId,
        port: MQTT_CONFIG.port,
    });

    mqttClient.on("connect", () => {
        console.log("MQTT已连接");
        // 订阅下行控制主题
        mqttClient.subscribe(MQTT_CONFIG.topic + "/command");
    });

    // 3. 处理下行控制指令（MQTT → Modbus写）
    mqttClient.on("message", async (topic, message) => {
        try {
            const cmd = JSON.parse(message.toString());
            console.log("收到控制指令:", cmd);
            if (cmd.type === "write_register") {
                await mbClient.writeRegister(cmd.address, cmd.value);
                console.log(`已写入寄存器${cmd.address}=${cmd.value}`);
            } else if (cmd.type === "write_coil") {
                await mbClient.writeCoil(cmd.address, cmd.value);
                console.log(`已写入线圈${cmd.address}=${cmd.value}`);
            }
        } catch (err) {
            console.error("控制指令执行失败:", err.message);
        }
    });

    // 4. 周期性采集并上报（Modbus → MQTT）
    async function pollAndPublish() {
        try {
            const payload = {
                deviceId: "device001",
                timestamp: new Date().toISOString(),
                data: {},
            };

            for (const point of DATA_POINTS) {
                const result = await mbClient.readHoldingRegisters(point.address, 1);
                let value = result.data[0];
                if (point.scale) value = value * point.scale;
                payload.data[point.name] = {
                    value: Number(value.toFixed(3)),
                    unit: point.unit,
                };
            }

            // 发布到MQTT
            mqttClient.publish(
                MQTT_CONFIG.topic + "/data",
                JSON.stringify(payload),
                { qos: 1 }
            );
            console.log("已上报:", JSON.stringify(payload.data));
        } catch (err) {
            console.error("采集上报失败:", err.message);
        }
    }

    // 启动采集
    setInterval(pollAndPublish, POLL_INTERVAL);
    pollAndPublish();  // 立即执行一次
}

main().catch(console.error);

九、错误处理与重连机制