代码收藏家技术教程 2023-02-03

简单易用的SpringBoot + Netty实现Modbus TCP主站的物联网云平台

本系列教程包括：
IOT云平台 simple（0）IOT云平台简介
 IOT云平台 simple（1）netty入门
 IOT云平台 simple（2）springboot入门
 IOT云平台 simple（3）springboot netty实现TCP Server
IOT云平台 simple（4）springboot netty实现简单的mqtt broker
IOT云平台 simple（5）springboot netty实现modbus TCP Master
IOT云平台 simple（6）springboot netty实现IOT云平台基本的架构（mqtt、Rabbitmq）

本章首先简单的介绍了modbus，然后利用springboot netty实现了简单的modbus TCP Master。

由于modbus是应答式的交互，这里通过HTTP请求触发springboot netty发送modbus TCP请求，网络调试工具收到请求后发送响应message。

这里：
modbus TCP Master：springboot netty
modbus TCP Slave：网络调试工具
postman：http触发springboot netty主动发送请求；

1 Modbus入门

Modbus在串行链路上分为Slave和Master、
Modbus协议使用的是主从通讯技术，即由主设备主动查询和操作设备。
Modbus Master：主控设备方；
Modbus Slave：从设备方。

Modbus协议有3种模式：
1）RTU
2）ASCII
3）TCP

1.1 Modbus-RTU

基于串口的Modbus-RTU 数据按照标准串口协议进行编码，是使用最广泛的一种Modbus协议，采用CRC-16_Modbus校验算法。

具体协议：

1.2 Modbus-ASCII

基于串口的Modbus-ASCII 所有数据都是ASCII格式，一个字节的原始数据需要两个字符来表示，效率低，采用LRC校验算法。

具体协议：

1.3 Modbus-TCP

基于网口的Modbus-TCP Modbus-TCP基于TCP/IP协议，占用502端口，数据帧主要包括两部分：MBAP（报文头）+PDU（帧结构），数据块与串行链路是一致的。

具体协议：

2 Modbus TCP master集成开发

Modbus TCP是运行在TCP/IP之上的应用层，所以master基本就是个TCP Server。
创建主要的类：
1） TCPServer
server类。
2 ）TCPServerChannelInitializer
server channel初始化的类
3）TCPServerStartListener：监听到springboot启动后，启动TCPServer。

TCPServerChannelInitializer中增加了编码解码器。

        socketChannel.pipeline().addLast("decoder", new TCPModbusResDecoder());
        socketChannel.pipeline().addLast("encoder", new TCPModbusReqEncoder());
        socketChannel.pipeline().addLast("tcpModbus", new TCPModbusResHandler());

2.1 定义message

定义message：

public class TCPModbusMessage {
    public MBAPHeader mbapHeader;
    public PduPayload pduPayload;
}

其中header：

public class MBAPHeader {
    //事务处理标识符 递增
    private short transactionId;
    //协议标识符 0x00 标识modbus协议
    private short protocolId;
    //长度,unitId + pdu长度
    private short length;
    //单元标识符,从机地址
    private byte unitId;

    public MBAPHeader(short transactionId, short protocolId, short length, byte unitId) {
        this.transactionId = transactionId;
    }

    public MBAPHeader() {

    }

    public int getTransactionId() {
        return transactionId;
    }

    public void setTransactionId(short transactionId) {
        this.transactionId = transactionId;
    }

    public int getProtocolId() {
        return protocolId;
    }

    public void setProtocolId(short protocolId) {
        this.protocolId = protocolId;
    }

    public int getLength() {
        return length;
    }

    public void setLength(short length) {
        this.length = length;
    }

    public short getUnitId() {
        return unitId;
    }

    public void setUnitId(byte unitId) {
        this.unitId = unitId;
    }

    public String toString() {
        return "transactionId:" + transactionId
                + ",protocolId:" + protocolId
                + ",length:" + length
                + ",unitId:" + unitId;

    }
}

其中pdu消息主体：

public class PduPayload {
    private short functionCode;
    private short dataLength;//数据字节的长度
    private byte[] data;

    public void setFunctionCode(short  code){
        functionCode = code;
    }
    public short getFunctionCode(){
        return functionCode;
    }

    public short getDataLength() {
        return dataLength;
    }

    public void setDataLength(short dataLength) {
        this.dataLength = dataLength;
    }

    public byte[] getData() {
        return data;
    }

    public void setData(byte[] data) {
        this.data = data;
    }

}

2.2 定义编码解码

2.2.1 解码

解码定义ByteToMessageDecoder：

public class TCPModbusResDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) {
        try {
            byteBuf.resetReaderIndex();
            int count = byteBuf.readableBytes();
            log.info("TCPModbusResDecoder decode:" + count);

            ByteBuf byteBuf1 = Unpooled.copiedBuffer(byteBuf);
            TCPModbusByteBufHolder tcpModbusByteBufHolder = new TCPModbusByteBufHolder(byteBuf1);
            list.add(tcpModbusByteBufHolder);

            byteBuf.clear();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

解码定义SimpleChannelInboundHandler：

public class TCPModbusResHandler extends SimpleChannelInboundHandler<TCPModbusByteBufHolder> {
    public static final int HEADER_LENGTH = 8;// // transactionId(2) + protocolId(2) + length(2) + unitId(1)+ functionCode(1)

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("channelActive");
        Channel channel = ctx.channel();
        log.info("channelActive channelId:" + channel.id().asLongText());
        log.info("channelActive 终端:" + channel.remoteAddress());
        ChannelManager.addChannel(channel.remoteAddress().toString(), channel);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("channelInactive");
        Channel channel = ctx.channel();
        ChannelManager.removeChannel(channel.remoteAddress().toString());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TCPModbusByteBufHolder modbusByteBufHolder) {
        log.info("channelRead0");
        int totalLen = modbusByteBufHolder.content().readableBytes();
        log.info("channelRead0:" + totalLen);
        if (totalLen < HEADER_LENGTH) {
            log.info("not modbus TCP protocol:");
            return;
        }
        //header
        MBAPHeader mbapHeader = MBAPHeaderDecoder.decode(modbusByteBufHolder.content());
        log.info("mbapHeader:" + mbapHeader.toString());
        //pdu
        PduPayload pduPayload = new PduPayload();
        short functionCode = modbusByteBufHolder.content().readUnsignedByte();
        pduPayload.setFunctionCode(functionCode);
        log.info("functionCode:" + functionCode);
        int dataLength = 0;
        if (totalLen > HEADER_LENGTH) {
            dataLength = totalLen - HEADER_LENGTH;
        }
        pduPayload.setDataLength((short) dataLength);
        log.info("dataLength:" + dataLength);
        byte[] data = new byte[dataLength];
        modbusByteBufHolder.content().readBytes(data);
        pduPayload.setData(data);

        RecMessageStrategy messageStrategy = RecMessageStrategyManager.getMessageStrategy(functionCode);
        if (messageStrategy != null) {
            messageStrategy.recMessage(channelHandlerContext.channel(), mbapHeader, pduPayload);
        } else {
            log.info("not support function code...");
        }

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        log.info("channelReadComplete");
        ctx.flush();
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        log.info("handlerRemoved");
        ctx.close();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("exceptionCaught");
        Channel channel = ctx.channel();
        ChannelManager.removeChannel(channel.remoteAddress().toString());
        cause.printStackTrace();
    }
}

这里针对Message接收处理定义了一个策略：

public interface RecMessageStrategy {
    void recMessage(Channel channel, MBAPHeader mbapHeader, PduPayload pduPayload);
}

增加了不同类型消息策略的管理：

public class RecMessageStrategyManager {

    //根据消息类型获取对应的策略类
    public static RecMessageStrategy getMessageStrategy(int functionCode){
        switch (functionCode){
            case FunctionCodeConstants
                    .ReadCoils:
                return new ReadCoilsResMessageStrategy();
            case FunctionCodeConstants
                    .ReadDiscreteInputs:
                return new ReadDiscreteInputsResMessageStrategy();
            case FunctionCodeConstants
                    .ReadHoldingRegisters:
                return new ReadHoldingRegistersResMessageStrategy();
            case FunctionCodeConstants
                    .ReadInputRegisters:
                return new ReadInputRegistersResMessageStrategy();
            case FunctionCodeConstants
                    .WriteSingleCoil:
                return new WriteSingleCoilResMessageStrategy();
            case FunctionCodeConstants
                    .WriteSingleRegister:
                return new WriteSingleRegisterResMessageStrategy();
            case FunctionCodeConstants
                    .WriteMultipleCoils:
                return new WriteMultipleCoilsResMessageStrategy();
            case FunctionCodeConstants
                    .WriteMultipleRegisters:
                return new WriteMultipleRegistersResMessageStrategy();
            default:
                return null;
        }
    }

}

2.2.2 编码

定义编码器：

public class TCPModbusReqEncoder extends MessageToByteEncoder<TCPModbusMessage> {

    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, TCPModbusMessage tcpModbusMessage, ByteBuf byteBuf) throws Exception {
        log.info("-----------TCPModbusReqEncoder encode begin------------");
        //header
        MBAPHeaderEncoder.encode(byteBuf, tcpModbusMessage.mbapHeader);
        log.info("header:"+tcpModbusMessage.mbapHeader.toString());

        //functionCode
        int functionCode = tcpModbusMessage.pduPayload.getFunctionCode();
        log.info("functionCode:"+functionCode);

        SendMessageStrategy sendMessageStrategy = new SendMessageStrategy();
        sendMessageStrategy.sendMessage(byteBuf,tcpModbusMessage.pduPayload);
        log.info("data:"+ ByteUtil.bytesToHexString(tcpModbusMessage.pduPayload.getData()));
        log.info("-----------TCPModbusReqEncoder encode end------------");
    }
}

这里定义了一个message发送的策略：

public class SendMessageStrategy {

    public ByteBuf sendMessage(ByteBuf byteBuf, PduPayload pduPayload) {
        log.info("SendMessageStrategy ");
        byteBuf.writeByte(pduPayload.getFunctionCode());
        byteBuf.writeBytes(pduPayload.getData());
        return byteBuf;
    }
}