背景

由于當(dāng)時項目周期趕，引入了一個PLC4X組件，上手快。接下來就是使用這個組件遇到的一些問題：

• 關(guān)閉連接NioEventLoop沒有釋放導(dǎo)致oom
• 設(shè)計思想是一個設(shè)備一個連接，而不是一個網(wǎng)關(guān)一個連接
• 連接斷開后客戶端無從感知
  前兩個問題解決方案參考上篇文章，最后一個問題雖然可以通過isConnect()方法獲取到狀態(tài)，但是連接斷開后這個狀態(tài)并沒有更新，只能代碼實現(xiàn)失敗重連。
  所以為了解決以上問題，我打算重新封裝一個Modbus組件。

步驟

代碼如下所示，目前只分享modbus-core相關(guān)的代碼。

• modbus-core：實現(xiàn)設(shè)備讀寫指令的下發(fā)以及應(yīng)答。
• modbus-app：實現(xiàn)通用的可靈活配置的modbus設(shè)備接入層，通過更新配置信息即可快速引入新設(shè)備，無需手寫代碼重啟應(yīng)用。

為了快速實現(xiàn)modbus組件封裝，這里引入了Vertx框架(基于事件+異步)官網(wǎng)鏈接,而不是原生的Netty框架。

引入架包

<!-- 目前我這里引入最新的版本(4.4.4) -->
<dependency>
     <groupId>io.vertx</groupId>
     <artifactId>vertx-core</artifactId>
     <version>${vertx.version}</version>
 </dependency>

工具類

ByteUtil

package com.bho.modbus.utils;

import java.nio.ByteBuffer;

public class ByteUtil {

    /**
     * 字節(jié)數(shù)組轉(zhuǎn)字符串
     * @param bytes
     * @return
     */
    public static  String bytesToHexString(byte[] bytes) {
        StringBuffer sb = new StringBuffer(bytes.length);
        String sTemp;
        for (int i = 0; i < bytes.length; i++) {
            sTemp = Integer.toHexString(0xFF & bytes[i]);
            if (sTemp.length() < 2) {
                sb.append(0);
            }
            sb.append(sTemp.toUpperCase());
        }
        return sb.toString();
    }

    /**
     * int整型轉(zhuǎn)字節(jié)數(shù)組
     * @param data
     * @param offset
     * @param len
     * @return
     */
    public static byte[] intToBytes(int data, int offset, int len) {
        ByteBuffer buffer = ByteBuffer.allocate(4);
        buffer.putInt(data);
        byte[] bytes = buffer.array();
        if (len - offset == 4) {
            return bytes;
        }
        byte[] dest = new byte[len];
        System.arraycopy(bytes, offset, dest, 0, len);
        return dest;
    }

    /**
     * 字節(jié)數(shù)組轉(zhuǎn)int整型
     * @param bytes
     * @param offset
     * @param len
     * @return
     */
    public static int bytesToInt(byte[] bytes, int offset, int len) {
        ByteBuffer buffer = ByteBuffer.allocate(4);
        for (int i = len; i < 4; i ++) {
            buffer.put((byte) 0x00);
        }
        for (int i = offset; i < offset + len; i++) {
            buffer.put(bytes[i]);
        }
        buffer.flip();
        return buffer.getInt();
    }



}

Crc16

package com.bho.modbus.utils;


public class Crc16 {

    /**
     * 獲取CRC16校驗碼
     * @param arr_buff
     * @return
     */
    public static byte[] getCrc16(byte[] arr_buff) {
        int len = arr_buff.length;

        // 預(yù)置 1 個 16 位的寄存器為十六進制FFFF, 稱此寄存器為 CRC寄存器。
        int crc = 0xFFFF;
        int i, j;
        for (i = 0; i < len; i++) {
            // 把第一個 8 位二進制數(shù)據(jù) 與 16 位的 CRC寄存器的低 8 位相異或, 把結(jié)果放于 CRC寄存器
            crc = ((crc & 0xFF00) | (crc & 0x00FF) ^ (arr_buff[i] & 0xFF));
            for (j = 0; j < 8; j++) {
                // 把 CRC 寄存器的內(nèi)容右移一位( 朝低位)用 0 填補最高位, 并檢查右移后的移出位
                if ((crc & 0x0001) > 0) {
                    // 如果移出位為 1, CRC寄存器與多項式A001進行異或
                    crc = crc >> 1;
                    crc = crc ^ 0xA001;
                } else
                    // 如果移出位為 0,再次右移一位
                    crc = crc >> 1;
            }
        }
        return intToBytes(crc);
    }


    private static byte[] intToBytes(int value) {
        byte[] src = new byte[2];
        src[1] = (byte) ((value >> 8) & 0xFF);
        src[0] = (byte) (value & 0xFF);
        return src;
    }



}

實體類

ModbusMode

目前只實現(xiàn)了以下兩種通信方式，可根據(jù)自己需求加入其它通信方式。

package com.bho.modbus.model;

import com.bho.modbus.utils.ByteUtil;
import com.bho.modbus.utils.Crc16;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.log4j.Log4j2;

import java.nio.ByteOrder;

@Log4j2
public enum ModbusMode {


    /**
     * 【事務(wù)ID(2) + 協(xié)議標(biāo)識(2) + 數(shù)據(jù)長度(2)】 + 從機地址(1) + 功能碼(1) + 數(shù)據(jù)區(qū)(N)
     */
    TCP,
    /**
     * 從機地址(1) + 功能碼(1) + 數(shù)據(jù)區(qū)(N) + 【校驗碼(2)】
     *
     */
    RTU,
    ;
    public ByteToMessageDecoder getDecoder() {
        if (this == ModbusMode.TCP) {
            return new LengthFieldBasedFrameDecoder(ByteOrder.BIG_ENDIAN, 65536, 4,
                    2, 0, 6, true);
        }
        if (this == ModbusMode.RTU){
            return new LengthFieldBasedFrameDecoder(ByteOrder.BIG_ENDIAN, 65536, 2,
                    1, 2, 0, true);
        }
        return null;
    }

    public byte[] readData(byte[] bytes) {
        int len = bytes.length;
        if (this == ModbusMode.RTU) {
            byte[] tempArr = new byte[len - 2];
            System.arraycopy(bytes, 0, tempArr, 0, tempArr.length);
            byte[] crc16 = Crc16.getCrc16(tempArr);
            if (crc16[0] != bytes[len -2] || crc16[1] != bytes[len - 1]) {
                log.error("Modbus receive illegal data:{}", ByteUtil.bytesToHexString(bytes));
                return null;
            }
            if (log.isDebugEnabled()) {
                log.debug("read data:{}", ByteUtil.bytesToHexString(tempArr));
            }
            return tempArr;
        }
        if (this == ModbusMode.TCP) {
            if (log.isDebugEnabled()) {
                log.debug("read data:{}", ByteUtil.bytesToHexString(bytes));
            }
            return bytes;
        }
        return null;
    }

    public byte[] writeData(byte[] bytes) {
        if (log.isDebugEnabled()) {
            log.debug("write data:{}",ByteUtil.bytesToHexString(bytes));
        }
        int len = bytes.length;
        if (this == ModbusMode.RTU) {
            byte[] crc16 = Crc16.getCrc16(bytes);
            byte[] tempArr = new byte[len + 2];
            System.arraycopy(bytes, 0, tempArr, 0, len);
            tempArr[len] = crc16[0];
            tempArr[len + 1] = crc16[1];
            return tempArr;
        }
        if (this == ModbusMode.TCP) {
            byte[] tempArr = new byte[len + 6];
            tempArr[1] = 0x01;
            byte[] lenBytes = ByteUtil.intToBytes(len, 2, 2);
            tempArr[4] = lenBytes[0];
            tempArr[5] = lenBytes[1];
            System.arraycopy(bytes, 0, tempArr, 6, len);
            return tempArr;
        }
        return null;
    }


}

ModbusFunc

功能碼

package com.bho.modbus.model;


/**
 * Modbus常見功能碼
 */
public enum ModbusFunc {

    /**
     * 錯誤代碼
     * 01：非法的功能碼
     * 02：非法的寄存器地址
     * 03：非法的數(shù)據(jù)值
     * 04：從機故障
     */


    /**
     * 請求:
     *      功能代碼：1字節(jié) 0x01
     *      起始地址：2字節(jié) 0x0000-0xffff
     *      線圈數(shù)量：2字節(jié) 0x0001-0x07d0(2000)
     *
     * 正確響應(yīng):
     *      功能代碼：1字節(jié) 0x01
     *        字節(jié)數(shù)：1字節(jié) N(讀線圈個數(shù)/8，余數(shù)不為0則加1)
     *      線圈狀態(tài)：N字節(jié)
     *
     * 錯誤響應(yīng):
     *      功能代碼：1字節(jié) 0x81
     *      錯誤代碼：1字節(jié) 0x01-0x04
     */
    READ_COILS((byte)0x01),//讀連續(xù)線圈狀態(tài)
    READ_DISCRETE_COILS((byte)0x02),//讀離散線圈狀態(tài) 同上


    /**
     * 請求:
     *       功能代碼：1字節(jié) 0x03
     *       起始地址：2字節(jié) 0x0000-0xffff
     *     寄存器數(shù)量：2字節(jié) 0x0001-0x007d(125)
     *
     * 正確響應(yīng):
     *       功能代碼：1字節(jié) 0x03
     *         字節(jié)數(shù)：1字節(jié) 2N(N為寄存器數(shù)量)
     *      寄存器數(shù)量：2N字節(jié)
     *
     * 錯誤響應(yīng):
     *      功能代碼：1字節(jié) 0x83
     *      錯誤代碼：1字節(jié) 0x01-0x04
     */
    READ_HOLDING_REGISTERS((byte)0x03),//讀保持寄存器值
    READ_INPUT_REGISTERS((byte)0x04),//讀輸入寄存器值 同上

    /**
     * 請求:
     *      功能代碼：1字節(jié) 0x05
     *      起始地址：2字節(jié) 0x0000-0xffff
     *      線圈狀態(tài)：2字節(jié) 0x0000/0xff00
     *
     * 正確響應(yīng):
     *      功能代碼：1字節(jié) 0x05
     *      起始地址：2字節(jié) 0x0000-0xffff
     *      線圈狀態(tài)：2字節(jié) 0x0000/0xff00
     *
     * 錯誤響應(yīng):
     *      功能代碼：1字節(jié) 0x85
     *      錯誤代碼：1字節(jié) 0x01-0x04
     */
    WRITE_SINGLE_COILS((byte)0x05),//寫單個線圈
    /**
     * 請求:
     *      功能代碼：1字節(jié) 0x06
     *      起始地址：2字節(jié) 0x0000-0xffff
     *      寄存器值：2字節(jié) 0x0000-0xffff
     *
     * 正確響應(yīng):
     *      功能代碼：1字節(jié) 0x06
     *      起始地址：2字節(jié) 0x0000-0xffff
     *      寄存器值：2字節(jié) 0x0000-0xffff
     *
     * 錯誤響應(yīng):
     *      功能代碼：1字節(jié) 0x86
     *      錯誤代碼：1字節(jié) 0x01-0x04
     */
    WRITE_SINGLE_HOLDING_REGISTERS((byte)0x06),//寫單個保持寄存器
    /**
     * 請求:
     *      功能代碼：1字節(jié) 0x10
     *      起始地址：2字節(jié) 0x0000-0xffff
     * 寫入寄存器個數(shù)：2字節(jié) 0x0001-0x007b(123)
     *    寫入字節(jié)數(shù)：1字節(jié) 2N(N為寄存器個數(shù))
     *     寄存器值：2N字節(jié) 0x0000-0xffff
     *
     * 正確響應(yīng):
     *      功能代碼：1字節(jié) 0x10
     *      起始地址：2字節(jié) 0x0000-0xffff
     * 寫入寄存器個數(shù)：2字節(jié) 0x0001-0x007b(123)
     *
     * 錯誤響應(yīng):
     *      功能代碼：1字節(jié) 0x90
     *      錯誤代碼：1字節(jié) 0x01-0x04
     */
    WRITE_MULTI_HOLDING_REGISTERS((byte)0x10),//寫多個保持寄存器
    /**
     * 請求:
     *      功能代碼：1字節(jié) 0x0F
     *      起始地址：2字節(jié) 0x0000-0xffff
     *   寫入線圈個數(shù)：2字節(jié) 0x0001-0x07b0(1968)
     *    寫入字節(jié)數(shù)：1字節(jié) N(N為線圈個數(shù)/8,余數(shù)不為0則加1)
     *     線圈狀態(tài)：N字節(jié)
     *
     * 正確響應(yīng):
     *      功能代碼：1字節(jié) 0x0F
     *      起始地址：2字節(jié) 0x0000-0xffff
     *   寫入線圈個數(shù)：2字節(jié) 0x0001-0x07b0(1968)
     *
     * 錯誤響應(yīng):
     *      功能代碼：1字節(jié) 0x8F
     *      錯誤代碼：1字節(jié) 0x01-0x04
     */
    WRITE_MULTI_COILS((byte)0x0F),//寫多個線圈

    ;

    private byte func;

    ModbusFunc(byte func) {
        this.func = func;
    }



    public byte getFunc() {
        return func;
    }
}

ModbusParamConfig

下發(fā)指令參數(shù)配置信息

package com.bho.modbus.model;


import lombok.Data;


@Data
public class ModbusParamConfig {

    private RegisterType registerType;//寄存器類型
    private int registerAddress;//寄存器地址
    private String name;//指標(biāo)名稱
    private DataType dataType;//指標(biāo)數(shù)據(jù)類型
    private int numberSplit;//(除)倍數(shù)

    public enum RegisterType {
        COIL,
        HOLDING_REGISTER,
        INPUT_REGISTER;
    }

    public enum DataType {
        BOOL,
        FLOAT,
        INT;
    }

}

SendCmdTask

下發(fā)指令任務(wù)

package com.bho.modbus.model;

import com.alibaba.fastjson.JSONObject;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import lombok.Data;

import java.util.List;


@Data
public class SendCmdTask {


    private List<ModbusParamConfig> paramConfigs;//參數(shù)列表
    private JSONObject reqParam;//請求參數(shù) 寫數(shù)據(jù)必填
    private Boolean isWrite;//是否是寫數(shù)據(jù)
    private Integer slaveId;//從機ID
    private Integer reqTimeout;//請求超時時間(秒)
    private Promise<JSONObject> promise;
    private Long timerId;


    public SendCmdTask(Vertx vertx, List<ModbusParamConfig> paramConfigs, JSONObject reqParam, Boolean isWrite, Integer slaveId, Integer reqTimeout) {
        this.paramConfigs = paramConfigs;
        this.reqParam = reqParam;
        this.isWrite = isWrite;
        this.slaveId = slaveId;
        this.reqTimeout = Math.max(reqTimeout, 5);
        Promise<JSONObject> promise = Promise.promise();
        this.promise = promise;
        this.timerId = vertx.setTimer(reqTimeout * 1000, hh -> promise.tryFail("Request timeout"));
    }
}

核心類

package com.bho.modbus.core;

import com.alibaba.fastjson.JSONObject;
import com.bho.modbus.model.SendCmdTask;
import com.bho.modbus.model.ModbusFunc;
import com.bho.modbus.model.ModbusMode;
import com.bho.modbus.model.ModbusParamConfig;

import com.bho.modbus.utils.ByteUtil;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.buffer.impl.BufferImpl;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;
import io.vertx.core.net.impl.NetSocketImpl;
import lombok.extern.log4j.Log4j2;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

@Log4j2
public class ModbusConnection {
    private String ip;//從機IP
    private Integer port;//從機端口
    private AtomicBoolean isAlive;//從機是否在線
    private ModbusMode mode;//通訊模式
    private NetSocket netSocket;//客戶端連接
    private boolean isInitiativeClose;//是否是主動關(guān)閉連接
    private Long failRetryTimerId;//失敗重試定時器ID
    private Integer failRetryIntervalSecond;//連接斷開后重連間隔時間
    private Integer reqTimeoutSecond = 1;//請求超時時間
    private Long queueTimerId;//隊列定時器
    private ConcurrentLinkedQueue<SendCmdTask> writeQueue;//寫隊列 優(yōu)先寫
    private ConcurrentLinkedQueue<SendCmdTask> readQueue;//讀隊列
    private Map<String, Promise<byte[]>> promiseMap;

    private Vertx vertx;

    public ModbusConnection(Vertx vertx, String ip, Integer port, Integer failRetryIntervalSecond, ModbusMode mode) {
        this.vertx = vertx;
        this.ip = ip;
        this.port = port;
        this.failRetryIntervalSecond = failRetryIntervalSecond;
        this.mode = mode;
        this.isAlive = new AtomicBoolean(false);
        this.writeQueue = new ConcurrentLinkedQueue<>();
        this.readQueue = new ConcurrentLinkedQueue<>();
        this.promiseMap = new ConcurrentHashMap<>();
        consumerTaskQueue(true);
    }

    /**
     * 建立連接
     * @return
     */
    public Future<Boolean> connect(){
        NetClient netClient = vertx.createNetClient();
        return vertx.executeBlocking(b -> {
            netClient.connect(port, ip)
                    .onSuccess(socket -> {
                        log.info("Modbus connect success, ip:{}, port:{}", ip, port);
                        netSocket  = socket;
                        isAlive.set(true);
                        b.tryComplete(true);
                        NetSocketImpl netSocketImpl = (NetSocketImpl) socket;
                        netSocketImpl.channelHandlerContext().pipeline().addFirst(mode.getDecoder());
                        socket.handler(buf -> {
                            byte[] bytes = mode.readData(buf.getBytes());
                            if (bytes == null) {
                                return;
                            }
                            int slaveId = ByteUtil.bytesToInt(bytes, 0, 1);
                            int funcNo = ByteUtil.bytesToInt(bytes, 1, 1);
                            int errFuncNo = funcNo - 128;
                            String key = String.format("%s_%s", slaveId, funcNo);
                            String errKey = String.format("%s_%s", slaveId, errFuncNo);
                            if (promiseMap.containsKey(key)) {
                                Promise<byte[]> promise = promiseMap.get(key);
                                byte[] content = new byte[bytes.length - 2];
                                System.arraycopy(bytes, 2, content, 0, content.length);
                                promise.tryComplete(content);
                            } else if (promiseMap.containsKey(errKey)) {
                                Promise<byte[]> promise = promiseMap.get(errKey);
                                int data = ByteUtil.bytesToInt(bytes, 2, 1);
                                switch (data) {
                                    case 1:
                                        promise.tryFail("Illegal function code");
                                        break;
                                    case 2:
                                        promise.tryFail("Illegal register address");
                                        break;
                                    case 3:
                                        promise.tryFail("Illegal data value");
                                        break;
                                    case 4:
                                        promise.tryFail("Slave fault");
                                        break;
                                }
                            }
                        });
                        socket.closeHandler(h -> {
                            if (!isInitiativeClose) {
                                log.error("Modbus connect close, ip:{}, port:{}", ip, port);
                                failRetryTimerId = vertx.setTimer(failRetryIntervalSecond * 1000, hh -> connect());
                            } else {
                                log.info("Modbus connect close, ip:{}, port:{}", ip, port);
                            }
                        });
                    }).onFailure(err -> {
                        log.error("Modbus connect fail, ip:{}, port:{}, msg:{}", ip, port, err.getMessage());
                        isAlive.set(false);
                        b.fail(err.getMessage());
                        failRetryTimerId = vertx.setTimer(failRetryIntervalSecond * 1000, h -> connect());
                    });
        });
    }

    /**
     * 是否在線
     * @return
     */
    public boolean isActive() {
        return isAlive.get();
    }

    /**
     * 斷開連接
     */
    public void close() {
        isInitiativeClose = true;
        if (failRetryTimerId != null) {
            vertx.cancelTimer(failRetryTimerId);
        }
        if (queueTimerId != null) {
            vertx.cancelTimer(queueTimerId);
        }
        if (netSocket != null) {
            netSocket.close();
        }
    }

    /**
     * 下發(fā)讀寫任務(wù)（串行 優(yōu)先寫任務(wù)）
     * 若并行可直接調(diào)用executeTask執(zhí)行任務(wù)，無需排隊等候一個個消費任務(wù)
     * @param task 讀寫任務(wù)
     * @return
     */
    public Promise<JSONObject> offerTask(SendCmdTask task) {
        if (task.getIsWrite()) {
            writeQueue.offer(task);
        } else {
            readQueue.offer(task);
        }
        return task.getPromise();
    }

    /**
     * 消費任務(wù)隊列 500毫秒輪詢一次 優(yōu)先消費寫任務(wù)
     * @param delayFlag
     */
    private void consumerTaskQueue(boolean delayFlag){
        if(delayFlag){
            queueTimerId = vertx.setTimer(500,id->{
                consumerTaskQueue(false);
            });
            return;
        }
        if(writeQueue.isEmpty() && readQueue.isEmpty()){
            consumerTaskQueue(true);
            return;
        }
        if(!writeQueue.isEmpty()){
            SendCmdTask sendCmdTask = writeQueue.poll();
            sendCmdTask.getPromise().future().onComplete(h->{
                consumerTaskQueue(false);
            });
            executeTask(sendCmdTask);
            return;
        }
        if(!readQueue.isEmpty()){
            SendCmdTask sendCmdTask = readQueue.poll();
            sendCmdTask.getPromise().future().onComplete(h->{
                consumerTaskQueue(false);
            });
            executeTask(sendCmdTask);
        }
    }

    private Future<Void> executeTask(SendCmdTask sendCmdTask){
        vertx.cancelTimer(sendCmdTask.getTimerId());
        Future<JSONObject> future;
        if (sendCmdTask.getIsWrite()) {
            future = executeWrite(sendCmdTask.getReqParam(), sendCmdTask.getParamConfigs(), sendCmdTask.getSlaveId());
        } else {
            future = executeQuery(sendCmdTask.getParamConfigs(), sendCmdTask.getSlaveId());
        }
        return future.onSuccess(res -> sendCmdTask.getPromise().tryComplete(res))
                .onFailure(err -> sendCmdTask.getPromise().tryFail(err)).map(o -> null);
    }

    /**
     * 寫數(shù)據(jù)
     * @param reqParam 下發(fā)參數(shù)
     * @param paramConfigs 參數(shù)配置列表
     * @param slaveId 從機ID
     * @return
     */
    private Future<JSONObject> executeWrite(JSONObject reqParam, List<ModbusParamConfig> paramConfigs, Integer slaveId) {
        if (!isActive()) {
            return Future.failedFuture("Gateway offline");
        }
        boolean isMerge = isMergeSendCmd(paramConfigs);
        if (isMerge) {
            int registerAddress = paramConfigs.get(0).getRegisterAddress();
            ModbusParamConfig.RegisterType registerType = paramConfigs.get(0).getRegisterType();
            Promise<byte[]> promise = Promise.promise();
            List<String> keyList = paramConfigs.stream().map(ModbusParamConfig::getName).collect(Collectors.toList());
            return vertx.executeBlocking(h -> {
                Buffer buffer = getWriteCmd(registerAddress, slaveId, reqParam, keyList, registerType, promise);
                netSocket.write(buffer);
                promise.future().onSuccess(buf -> {
                    h.complete(reqParam);
                }).onFailure(err -> {
                    log.error("Modbus executeWrite fail, ip:{}, port:{}, slaveId:{}, msg:{}", ip, port, slaveId, err.getMessage());
                    h.tryFail(err.getMessage());
                });
            });
        }

        List<Future<Object>> futures = new ArrayList<>();
        Future blockingFuture = Future.succeededFuture();
        for (int i = 0; i < paramConfigs.size(); i++) {
            ModbusParamConfig paramConfig = paramConfigs.get(i);
            ModbusParamConfig.RegisterType registerType = paramConfig.getRegisterType();
            Promise<byte[]> promise = Promise.promise();
            blockingFuture = blockingFuture.compose(suc -> singleExecuteWrite(slaveId, reqParam, promise, registerType, paramConfig),
                    err -> singleExecuteWrite(slaveId, reqParam, promise, registerType, paramConfig));
            futures.add(blockingFuture);
        }
        return commonReplyResult(futures, paramConfigs);

    }

    private Future<Object> singleExecuteWrite(int slaveId, JSONObject reqParam, Promise<byte[]> promise, ModbusParamConfig.RegisterType registerType, ModbusParamConfig paramConfig) {
        return vertx.executeBlocking(h -> {
            Buffer buffer = getWriteCmd(paramConfig.getRegisterAddress(), slaveId, reqParam, Arrays.asList(paramConfig.getName()), registerType, promise);
            netSocket.write(buffer);
            promise.future().onSuccess(buf -> {
                h.tryComplete(reqParam.get(paramConfig.getName()));
            }).onFailure(err -> {
                log.error("Modbus executeWrite fail, ip:{}, port:{}, slaveId:{}, key:{}, msg:{}",
                        ip, port, slaveId, paramConfig.getName(), err.getMessage());
                h.tryFail(err.getMessage());
            });
        });
    }

    /**
     * 讀數(shù)據(jù)
     * @param paramConfigs 參數(shù)配置列表
     * @param slaveId 從機ID
     * @return
     */
    private Future<JSONObject> executeQuery(List<ModbusParamConfig> paramConfigs, Integer slaveId) {
        if (!isActive()) {
            return Future.failedFuture("Gateway offline");
        }
        boolean isMerge = isMergeSendCmd(paramConfigs);
        if (isMerge) {
            int registerAddress = paramConfigs.get(0).getRegisterAddress();
            ModbusParamConfig.RegisterType registerType = paramConfigs.get(0).getRegisterType();
            int num = paramConfigs.size();
            Promise<byte[]> promise = Promise.promise();
            Buffer buffer = getQueryCmd(registerAddress, num, slaveId, registerType, promise);
            return vertx.executeBlocking(h -> {
                netSocket.write(buffer);
                promise.future().onSuccess(buf -> {
                    JSONObject jsonObject = new JSONObject();
                    for (int i = 0; i < paramConfigs.size(); i++) {
                        ModbusParamConfig paramConfig = paramConfigs.get(i);
                        switch (registerType) {
                            case COIL:
                                Integer pow = Double.valueOf(Math.pow(2, i % 8)).intValue();
                                jsonObject.put(paramConfig.getName(), (pow & buf[i / 8 + 1]) == pow);
                                break;
                            case INPUT_REGISTER:
                            case HOLDING_REGISTER:
                                jsonObject.put(paramConfig.getName(), getValue(ByteUtil.bytesToInt(buf, i * 2 + 1, 2), paramConfig.getNumberSplit(), paramConfig.getDataType()));
                                break;
                        }
                    }
                    h.complete(jsonObject);
                }).onFailure(err -> {
                    log.error("Modbus executeQuery fail, ip:{}, port:{}, slaveId:{}, msg:{}", ip, port, slaveId, err.getMessage());
                    h.tryFail(err.getMessage());
                });
            });
        }
        List<Future<Object>> futures = new ArrayList<>();
        Future blockingFuture = Future.succeededFuture();
        for (int i = 0; i < paramConfigs.size(); i++) {
            ModbusParamConfig paramConfig = paramConfigs.get(i);
            ModbusParamConfig.RegisterType registerType = paramConfig.getRegisterType();
            Promise<byte[]> promise = Promise.promise();
            blockingFuture = blockingFuture.compose(suc -> singleExecuteQuery(slaveId, promise, registerType, paramConfig),
                    err -> singleExecuteQuery(slaveId, promise, registerType, paramConfig));
            futures.add(blockingFuture);
        }
        return commonReplyResult(futures, paramConfigs);
    }

    private Future<Object> singleExecuteQuery(int slaveId, Promise<byte[]> promise, ModbusParamConfig.RegisterType registerType, ModbusParamConfig paramConfig) {
        return vertx.executeBlocking(h -> {
            Buffer buffer = getQueryCmd(paramConfig.getRegisterAddress(), 1, slaveId, paramConfig.getRegisterType(), promise);
            netSocket.write(buffer);
            promise.future().onSuccess(buf -> {
                switch (registerType) {
                    case COIL:
                        h.complete(Integer.valueOf(buf[1]) == 1);
                        break;
                    case INPUT_REGISTER:
                    case HOLDING_REGISTER:
                        h.complete(getValue(ByteUtil.bytesToInt(buf, 1, 2), paramConfig.getNumberSplit(), paramConfig.getDataType()));
                        break;
                }
            }).onFailure(err -> {
                log.error("Modbus executeQuery fail, ip:{}, port:{}, slaveId:{}, key:{}, msg:{}",
                        ip, port, slaveId, paramConfig.getName(), err.getMessage());
                h.tryFail(err.getMessage());
            });
        });
    }

    /**
     * 如果所有參數(shù)寄存器類型一致并且地址連續(xù) 則合并成一條命令下發(fā)
     * @param paramConfigs
     * @return 是否可以合并下發(fā)命令
     */
    private boolean isMergeSendCmd(List<ModbusParamConfig> paramConfigs) {
        if (paramConfigs.size() == 1) {
            return false;
        }
        int lastPos = paramConfigs.get(0).getRegisterAddress();
        ModbusParamConfig.RegisterType registerType = paramConfigs.get(0).getRegisterType();
        for (int i = 1; i < paramConfigs.size(); i++) {
            int curPos = paramConfigs.get(i).getRegisterAddress();
            if (curPos - lastPos != 1) {
                return false;
            }
            ModbusParamConfig.RegisterType curRegisterType = paramConfigs.get(i).getRegisterType();
            if (registerType != curRegisterType) {
                return false;
            }
            lastPos = curPos;
        }
        return true;
    }

    /**
     * 獲取查詢數(shù)據(jù)命令
     * @param startPos 查詢地址
     * @param num 查詢數(shù)量
     * @param slaveId 從機ID
     * @param registerType 寄存器類型
     * @param promise
     * @return
     */
    private Buffer getQueryCmd(int startPos, int num, int slaveId, ModbusParamConfig.RegisterType registerType, Promise<byte[]> promise) {
        byte[] bytes = new byte[6];
        bytes[0] = ByteUtil.intToBytes(slaveId, 3, 1)[0];
        switch (registerType) {
            case COIL:
                bytes[1] = ModbusFunc.READ_COILS.getFunc();
                break;
            case HOLDING_REGISTER:
                bytes[1] = ModbusFunc.READ_HOLDING_REGISTERS.getFunc();
                break;
            case INPUT_REGISTER:
                bytes[1] = ModbusFunc.READ_INPUT_REGISTERS.getFunc();
                break;
        }
        Integer func = ByteUtil.bytesToInt(bytes, 1, 1);
        String key = String.format("%s_%s", slaveId, func);
        byte[] startPosBytes = ByteUtil.intToBytes(startPos, 0, 4);
        bytes[2] = startPosBytes[2];
        bytes[3] = startPosBytes[3];
        byte[] numBytes = ByteUtil.intToBytes(num, 0, 4);
        bytes[4] = numBytes[2];
        bytes[5] = numBytes[3];
        Buffer buffer = new BufferImpl();
        buffer.appendBytes(mode.writeData(bytes));
        promiseMap.put(key, promise);
        long timeId = vertx.setTimer(reqTimeoutSecond * 1000, h -> promise.tryFail("Request timeout"));
        promise.future().onComplete(res -> {
            promiseMap.remove(key);
            vertx.cancelTimer(timeId);
        });
        return buffer;
    }

    /**
     * 獲取寫數(shù)據(jù)命令
     * @param startPos 查詢地址
     * @param slaveId 從機ID
     * @param reqParam 寫參數(shù)
     * @param keys  參數(shù)列表
     * @param registerType 寄存器類型
     * @param promise
     * @return
     */
    private Buffer getWriteCmd(int startPos, int slaveId, JSONObject reqParam,
                               List<String> keys, ModbusParamConfig.RegisterType registerType, Promise<byte[]> promise) {
        int len = keys.size() == 1 ? 6 : (registerType == ModbusParamConfig.RegisterType.HOLDING_REGISTER ?
                7 + keys.size() * 2 : 7 + Double.valueOf(Math.ceil(keys.size() / 8.0)).intValue());
        byte[] bytes = new byte[len];
        bytes[0] = ByteUtil.intToBytes(slaveId, 3, 1)[0];
        byte[] startPosBytes = ByteUtil.intToBytes(startPos, 0, 4);
        bytes[2] = startPosBytes[2];
        bytes[3] = startPosBytes[3];
        if (keys.size() == 1) {
            switch (registerType) {
                case COIL:
                    bytes[1] = ModbusFunc.WRITE_SINGLE_COILS.getFunc();
                    boolean value = reqParam.getBoolean(keys.get(0));
                    if (value) {
                        bytes[4] = (byte) 0xFF;
                    } else {
                        bytes[4] = 0x00;
                    }
                    bytes[5] = 0x00;
                    break;
                case HOLDING_REGISTER:
                    bytes[1] = ModbusFunc.WRITE_SINGLE_HOLDING_REGISTERS.getFunc();
                    byte[] dataArr = ByteUtil.intToBytes(reqParam.getInteger(keys.get(0)), 2, 2);
                    bytes[4] = dataArr[0];
                    bytes[5] = dataArr[1];
                    break;
            }
        } else {
            byte[] dataNum = ByteUtil.intToBytes(keys.size(), 2, 2);
            bytes[4] = dataNum[0];
            bytes[5] = dataNum[1];
            switch (registerType) {
                case COIL:
                    bytes[1] = ModbusFunc.WRITE_MULTI_COILS.getFunc();
                    int dataSize = Double.valueOf(Math.ceil(keys.size() / 8.0)).intValue();
                    bytes[6] = ByteUtil.intToBytes(dataSize, 3, 1)[0];
                    for (int i = 0; i < dataSize; i += 2) {
                        int sum = 0;
                        int startIndex = i * 8;
                        int endIndex = (i + 2) * 8;
                        endIndex = endIndex > keys.size() ? keys.size() : endIndex;
                        for (int j = startIndex; j < endIndex; j++) {
                            sum += Double.valueOf(Math.pow(2, j)).intValue() * (reqParam.getBoolean(keys.get(j)) ? 1 : 0);
                        }
                        byte[] sumArr = ByteUtil.intToBytes(sum, 2, 2);
                        if (i + 8 < keys.size()) {
                            bytes[i + 7] = sumArr[0];
                            bytes[i + 8] = sumArr[1];
                        } else {
                            bytes[i + 7] = sumArr[1];
                        }
                    }
                    break;
                case HOLDING_REGISTER:
                    bytes[1] = ModbusFunc.WRITE_MULTI_HOLDING_REGISTERS.getFunc();
                    bytes[6] = ByteUtil.intToBytes(keys.size() * 2, 3, 1)[0];
                    for (int i = 0; i < keys.size(); i++) {
                        String paramKey = keys.get(i);
                        Integer value = reqParam.getInteger(paramKey);
                        byte[] dataArr = ByteUtil.intToBytes(value, 2, 2);
                        bytes[i * 2 + 7] = dataArr[0];
                        bytes[i * 2 + 8] = dataArr[1];
                    }
                    break;
            }
        }
        Integer func = ByteUtil.bytesToInt(bytes, 1, 1);
        String key = String.format("%s_%s", slaveId, func);
        Buffer buffer = new BufferImpl();
        buffer.appendBytes(mode.writeData(bytes));
        promiseMap.put(key, promise);
        long timeId = vertx.setTimer(reqTimeoutSecond * 1000, h -> promise.tryFail("Request timeout"));
        promise.future().onComplete(res -> {
            promiseMap.remove(key);
            vertx.cancelTimer(timeId);
        });
        return buffer;
    }

    private Future<JSONObject> commonReplyResult(List<Future<Object >> futures, List<ModbusParamConfig> paramConfigs) {
        return vertx.executeBlocking(b -> {
            Future.join(futures).onComplete(h -> {
                JSONObject okJson = new JSONObject();
                JSONObject errJson = new JSONObject();
                for (int i = 0; i < paramConfigs.size(); i++) {
                    ModbusParamConfig paramConfig = paramConfigs.get(i);
                    Future<Object> objectFuture = futures.get(i);
                    if (objectFuture.succeeded()) {
                        okJson.put(paramConfig.getName(), objectFuture.result());
                    } else {
                        errJson.put(paramConfig.getName(), objectFuture.cause().getMessage());
                    }
                }
                if (okJson.size() > 0) {
                    b.tryComplete(okJson);
                } else {
                    b.tryFail(errJson.getString(paramConfigs.get(0).getName()));
                }
            });
        });
    }

    private Object getValue(int value, int numberSplit, ModbusParamConfig.DataType dataType) {
        if (numberSplit == 1) {
            return value;
        }
        Float temp = value * 1f / numberSplit;
        switch (dataType) {
            case INT :
                return Math.round(temp);
            case FLOAT:
                return temp;
        }
        return temp;
    }

}

測試

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bho.modbus.model.ModbusMode;
import com.bho.modbus.core.ModbusConnection;
import com.bho.modbus.model.ModbusParamConfig;
import com.bho.modbus.model.SendCmdTask;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.core.Vertx;
import lombok.extern.log4j.Log4j2;

import java.util.List;

@Log4j2
public class TestModbus {
    public static final String READ_DATA = "[" +
            "        {" +
            "          \"name\": \"a\"," +
            "          \"registerType\": \"HOLDING_REGISTER\"," +
            "          \"registerAddress\": 504," +
            "          \"dataType\": \"FLOAT\"," +
            "          \"numberSplit\": 10" +
            "        }," +
            "        {" +
            "          \"name\": \"b\"," +
            "          \"registerType\": \"HOLDING_REGISTER\"," +
            "          \"registerAddress\": 505," +
            "          \"dataType\": \"FLOAT\"," +
            "          \"numberSplit\": 10" +
            "        }," +
            "        {" +
            "          \"name\": \"c\"," +
            "          \"registerType\": \"HOLDING_REGISTER\"," +
            "          \"registerAddress\": 506," +
            "          \"dataType\": \"FLOAT\"," +
            "          \"numberSplit\": 10" +
            "        }," +
            "        {" +
            "          \"name\": \"d\"," +
            "          \"registerType\": \"HOLDING_REGISTER\"," +
            "          \"registerAddress\": 507," +
            "          \"dataType\": \"INT\"," +
            "          \"numberSplit\": 1" +
            "        }," +
            "        {" +
            "          \"name\": \"e\"," +
            "          \"registerType\": \"HOLDING_REGISTER\"," +
            "          \"registerAddress\": 508," +
            "          \"dataType\": \"INT\"," +
            "          \"numberSplit\": 1" +
            "        }]";

    private static final String WRITE_DATA = "[" +
            "        {" +
            "          \"name\": \"do0\"," +
            "          \"registerType\": \"COIL\"," +
            "          \"registerAddress\": 20," +
            "          \"dataType\": \"BOOL\"," +
            "          \"numberSplit\": 1" +
            "        }" +
            "        ,{" +
            "          \"name\": \"do1\"," +
            "          \"registerType\": \"COIL\"," +
            "          \"registerAddress\": 21," +
            "          \"dataType\": \"BOOL\"," +
            "          \"numberSplit\": 1" +
            "        }" +

            "]";
    public static void main(String[] args) {
        testReadData();
//        testWriteData();;


    }

    private static void testWriteData() {
        Vertx vertx = Vertx.vertx();
        ModbusConnection connection = new ModbusConnection(vertx,"127.0.0.1", 502, 30, ModbusMode.TCP);
        Future<Boolean> connectFuture = connection.connect();
        JSONObject reqParam = new JSONObject();
        reqParam.put("do0", false);
        reqParam.put("do1", false);
        List<ModbusParamConfig> modbusParamConfigs = JSONArray.parseArray(WRITE_DATA, ModbusParamConfig.class);
        connectFuture.onComplete(con -> {
            if (connectFuture.succeeded()) {
                SendCmdTask task = new SendCmdTask(vertx, modbusParamConfigs, null, false, 21, 10);
                Promise<JSONObject> promise = connection.offerTask(task);
                promise.future().onSuccess(suc -> {
                    log.info("read:"+suc);
                }).onFailure(err -> System.err.println(err.getMessage()));

                SendCmdTask task2 = new SendCmdTask(vertx, modbusParamConfigs, reqParam, true, 21, 10);
                Promise<JSONObject> promise2 = connection.offerTask(task2);
                promise2.future().onSuccess(suc -> {
                    log.info("write:"+suc);
                }).onFailure(err -> System.err.println(err.getMessage()));
            } else {
                System.err.println("gateway offline");
            }
        });

    }

    private static void testReadData() {
        Vertx vertx = Vertx.vertx();
        ModbusConnection connection = new ModbusConnection(vertx,"127.0.0.1", 502, 30, ModbusMode.TCP);
        Future<Boolean> connectFuture = connection.connect();
        List<ModbusParamConfig> modbusParamConfigs = JSONArray.parseArray(READ_DATA, ModbusParamConfig.class);
        connectFuture.onComplete(con -> {
            if (connection.isActive()) {
                SendCmdTask task = new SendCmdTask(vertx, modbusParamConfigs, null, false, 2, 10);
                Promise<JSONObject> promise = connection.offerTask(task);
                promise.future().onSuccess(suc -> {
                    log.info(suc);
                }).onFailure(err -> System.err.println(err.getMessage()));
            } else {
                System.err.println("gateway offline");
            }
        });

    }
}

運行結(jié)果如下：
其實這兩個讀寫示例如果是一個網(wǎng)關(guān)可以共用一個Modbus連接。

modbus-app配置參數(shù)

格式如下：

{
  "readable": {
    "devType01": {
      "ReportData": [
        {
          "name" : "xxx",
          "registerType" : "COIL",
          "registerAddress" : 1,
          "dataType" : "BOOL",
          "numberSplit" : 1
        }
      ]
    },
    "devType02": {
      "ReportData": [
        {
          "name" : "a",
          "registerType" : "HOLDING_REGISTER",
          "registerAddress" : 1,
          "dataType" : "INT",
          "numberSplit" : 1
        },
        {
          "name" : "b",
          "registerType" : "HOLDING_REGISTER",
          "registerAddress" : 2,
          "dataType" : "INT",
          "numberSplit" : 10
        },
        {
          "name": "c",
          "registerType": "",
          "dataType": "FLOAT",
          "mbScript": "(a*10000+b)/10"
        }
      ]
    }
  },
  "writable": {
    "devType01": {
      "Control": [
        {
          "name": "operation",
          "registerType": "COIL",
          "registerAddress": 21,
          "dataType": "BOOL",
          "numberSplit": 1
        }
      ]
    }
  },
  "readDataPeriods": [
    {
      "period" : 60,
      "deviceTypes": ["devType01"]
    },
    {
      "period" : 600,
      "deviceTypes": ["devType02","devType03"]
    }
  ]
}

具體怎么實現(xiàn)這邊就不過多講解了…

結(jié)束

不保證代碼正確，我這邊只是大概實現(xiàn)了一下，僅供參考。若有問題，請批評指出，我會虛心接受并積極修復(fù)問題。文章來源地址http://www.zghlxwxcb.cn/news/detail-704632.html

到了這里，關(guān)于Java實現(xiàn)Modbus讀寫數(shù)據(jù)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！