简介Java与Netty实现的电力104协议服务端源代码面向电力系统通信开发者及Java网络编程学习者。项目依据IEC 60870-5-104标准构建了完整的服务端通信框架覆盖RTU与SCADA调度中心之间的远程数据交换。核心代码涉及ASDU应用服务数据单元的解析与封装、控制信息字段处理、二进制及ASCII格式的编解码、报文校验与异常重传、超时与连接恢复机制等实现了从网络层到业务层的模块化解耦。压缩包共61个文件其中包含55个Java源文件另有properties配置文件用于参数调整、xml工程文件辅助项目构建、jar依赖包支持运行以及README说明帮助快速上手整体仅105KB结构清晰便于检索。已有134人学习浏览。借助此源码可深入理解Netty异步非阻塞I/O模型在工业协议通信中的落地方式学习104协议报文的分层解析与状态机设计同时掌握高并发连接管理、性能优化和容错策略适合具备Java基础并希望参与电力自动化系统开发的读者作为实战参考。1. Java实现电力104协议对接为什么我选了Netty自研服务最近在给一个地区调度做数据接入任务落到手里就是 Java实现电力104协议对接 netty服务源代码。一开始我也想直接用现成的开源协议栈翻了一圈发现两个问题一是版本停在老框架上二是一致性测试出问题时你根本无从下手。最后还是决定用Netty自己写主站服务端协议解析、链路状态机、编解码器全部自己控制。电力104协议IEC 60870-5-104本质上是运行在TCP长连接上的应用层帧协议Netty的事件驱动模型和ByteBuf正好把“粘包拆包”这个最烦人的问题变成管道里的一小段逻辑。这篇文章适合两类人一类是刚接手电力接入项目、需要快速跑通从站数据采集的Java后端另一类是已经被“链路正常但数据不刷新”折磨过、想搞明白104链路状态机的工程师。2. 先把104帧结构拆明白APCI与ASDU的字节布局2.1 I帧、S帧、U帧三种控制域决定链路怎么走IEC 60870-5-104把一帧数据切成两个部分APCI应用协议控制信息固定4字节和ASDU应用服务数据单元可变长。TCP流里看到的每一帧都是这样排布的启动字符长度L控制域APCIASDU0x681字节4字节L-4字节长度L是关键它表示从控制域第一个字节开始到帧尾的字节数也就是4加ASDU长度。最小L是4纯控制帧最大不能超过253因为L本身只有1字节。注意L不含启动字符0x68和L自己所以解码时一帧完整长度是2加L。控制域的前4个字节决定这帧属于哪种类型。I帧最常用承载遥测、遥信、遥控报文带发送序号和接收序号需要对方确认S帧只带接收序号用来确认收到哪些I帧不带ASDUU帧不带序号专门做链路控制典型命令是STARTDT启动数据传输、STOPDT停止数据传输、TESTFR测试帧也就是心跳探测。判断方式很简单控制域第一个字节的最低位bit0为0就是I帧为1再看具体值0x01是S帧0x07、0x0B、0x13、0x43这些就是U帧的各种命令。这里有个新手必踩的认知误区104协议虽然跑在TCP上但它自己维护了一套传输确认机制不是TCP确认了就完事。TCP只保证字节流到达104应用层还要检查“序号连续不连续、该确认的是不是确认了”。所以写业务前先把I帧的发送序号和接收序号怎么增减搞清楚否则后面联调时你会看到从站一直不回数据。2.2 ASDU里装着遥测、遥信和电度类型标识与信息体地址ASDU才是真正有业务含义的部分。常见的ASDU结构是类型标识可变结构限定词VSQ传送原因COT公共地址信息体1字节1字节2字节2字节N个信息体类型标识决定后面信息体怎么解释类型1是单点遥信类型13是带品质的短浮点遥测类型100是总召唤命令类型30是带时标的单点遥信。VSQ的低7位表示本帧包含几个信息体最高位表示信息体地址是否连续COT的低6位是传送原因比如1是周期、3是突发、6是激活、20是响应总召。公共地址是变电站站地址很多项目里一台主站接几十个站就是靠这个字段区分数据属于哪台从站。信息体地址是3字节低位在前排列。它不像HTTP URL有统一语义在104里遥测、遥信、电度分别落在哪个地址段完全看厂站的点表配置。我一般会把这部分做成配置映射收到信息体地址后查表转成内部测点名。千万别在代码里硬编码地址不同厂家、不同工程差距非常大。2.3 把帧结构映射成Java对象字段设计与字节顺序写104对接源码我习惯不引入太重的协议类框架直接用byte数组和ByteBuf操作自己封装一个帧对象。字节序这块必须从一开始就统一104的多字节字段是低字节在前Little-Endian信息体地址、公共地址、短浮点数都是这样。用Netty的ByteBuf时默认大端序反而容易踩坑所以我解析时全部用手工读字节再组装。public class Iec104Frame { public static final byte START_BYTE 0x68; private final byte[] ctrl new byte[4]; private byte[] asdu; public Iec104Frame() { } public Iec104Frame(byte[] ctrl, byte[] asdu) { System.arraycopy(ctrl, 0, this.ctrl, 0, 4); this.asdu asdu; } public static Iec104Frame buildIFrame(int sendSeq, int recvSeq, byte[] asdu) { Iec104Frame frame new Iec104Frame(); // 发送序号和接收序号都左移1位写入最低位留作帧类型标志 frame.ctrl[0] (byte) ((sendSeq 1) 0xFF); frame.ctrl[1] (byte) ((sendSeq 7) 0xFF); frame.ctrl[2] (byte) ((recvSeq 1) 0xFF); frame.ctrl[3] (byte) ((recvSeq 7) 0xFF); frame.asdu asdu; return frame; } public static Iec104Frame buildSFrame(int recvSeq) { Iec104Frame frame new Iec104Frame(); frame.ctrl[0] 0x01; frame.ctrl[1] 0x00; frame.ctrl[2] (byte) ((recvSeq 1) 0xFF); frame.ctrl[3] (byte) ((recvSeq 7) 0xFF); frame.asdu new byte[0]; return frame; } public static Iec104Frame buildUFrame(byte command) { Iec104Frame frame new Iec104Frame(); frame.ctrl[0] command; frame.ctrl[1] 0x00; frame.ctrl[2] 0x00; frame.ctrl[3] 0x00; frame.asdu new byte[0]; return frame; } public boolean isIFrame() { return (ctrl[0] 0x01) 0; } public int getSendSeq() { if (!isIFrame()) { throw new IllegalStateException(不是I帧); } return ((ctrl[0] 0xFF) 1) | ((ctrl[1] 0xFF) 7); } public int getRecvSeq() { return ((ctrl[2] 0xFF) 1) | ((ctrl[3] 0xFF) 7); } public boolean isSFrame() { return (ctrl[0] 0xFF) 0x01; } public byte[] getCtrl() { return ctrl; } public byte[] getAsdu() { return asdu; } }这个类把帧类型判断和序号读写收敛在一起后续编解码器只用它。注意buildIFrame里序号左移一位的写法这是104规约里序号的真实落盘方式序号本身加一条最低位标识所以收到0x02代表的发送序号是1不是2。如果你用别的开源实现一定要先确认它有没有做这一步移位否则联调时序号永远对不上。3. 用Netty搭一个104主站服务工程结构与链路建立3.1 Maven依赖与最少类清单先解决依赖。Netty推荐4.1.x用最普通的聚合包就够了dependency groupIdio.netty/groupId artifactIdnetty-all/artifactId version4.1.100.Final/version /dependency不要为了省依赖去引netty-handler单独包后面调试粘包时你会发现缺了ByteBuf相关工具类。工程结构上最少需要四个类服务启动类、ChannelInitializer、帧解码器、帧编码器加上链路状态Handler。我自己还会单独拆一个ASDU解析类避免Handler越来越胖。3.2 启动入口EventLoopGroup与ChannelInitializer怎么配104主站服务本质上是一个TCP服务端监听2404端口等着各从站连上来。Netty的启动参数不要照抄HTTP服务的套路104长连接的特点是连接数量不算大但每路连接消息持续不断所以worker线程数按CPU核数配别贪多。public final class Iec104Server { public static void main(String[] args) throws Exception { EventLoopGroup bossGroup new NioEventLoopGroup(1); EventLoopGroup workerGroup new NioEventLoopGroup( Math.max(2, Runtime.getRuntime().availableProcessors() * 2)); try { ServerBootstrap bootstrap new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 128) .childOption(ChannelOption.TCP_NODELAY, true) .childOption(ChannelOption.SO_KEEPALIVE, true) .childHandler(new ChannelInitializerSocketChannel() { Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(frameDecoder, new Iec104FrameDecoder()); ch.pipeline().addLast(frameEncoder, new Iec104FrameEncoder()); ch.pipeline().addLast(linkHandler, new Iec104LinkHandler()); } }); ChannelFuture future bootstrap.bind(2404).sync(); future.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } }SO_BACKLOG设成128足够104从站一般就几十个。TCP_NODELAY必须开104报文交互频繁Nagle算法会把小报文攒着不发直接导致链路握手看起来像卡死。SO_KEEPALIVE开的是TCP层保活但真正的心跳是应用层的TESTFR帧TCP保活只是兜底别指望它。3.3 握手时序U帧STARTDT怎么发、确认怎么收TCP连接建立后主站不能直接发遥测查询必须先走一遍104链路握手主站发STARTDT激活帧U帧0x07从站回STARTDT确认帧U帧0x0B确认后才允许收发I帧。很多第一次写的人把握手当成可有可无的步骤结果从站一律不回业务数据。链路状态机的四个状态是未连接、等待STARTDT确认、运行、停止。我把状态迁移放在Iec104LinkHandler里用AtomicReference保存状态避免多线程读脏。public class Iec104LinkHandler extends SimpleChannelInboundHandlerByteBuf { private enum LinkState { WAIT_START_CON, STARTED, STOPPED } private final AtomicReferenceLinkState state new AtomicReference(LinkState.WAIT_START_CON); private int sendSeq; private int recvSeq; private ScheduledFuture? testFrameTask; Override public void channelActive(ChannelHandlerContext ctx) throws Exception { // 连接建立立刻发STARTDT激活 Iec104Frame startdt Iec104Frame.buildUFrame((byte) 0x07); writeFrame(ctx, startdt); // T3超时后发TESTFR心跳这里先按20秒一轮 testFrameTask ctx.executor().scheduleWithFixedDelay(() - { if (state.get() LinkState.STARTED) { writeFrame(ctx, Iec104Frame.buildUFrame((byte) 0x43)); } }, 20, 20, TimeUnit.SECONDS); } Override protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception { // 复制成byte[]再解析避免ByteBuf被后续业务线程池释放 byte[] frameBytes new byte[msg.readableBytes()]; msg.readBytes(frameBytes); Iec104Frame frame parseFrame(frameBytes); if (!frame.isIFrame() !frame.isSFrame() frame.getCtrl()[0] (byte) 0x0B) { // STARTDT确认链路进入运行态接着发总召 if (state.compareAndSet(LinkState.WAIT_START_CON, LinkState.STARTED)) { sendGeneralCall(ctx); } } else if (frame.isIFrame()) { // 处理遥测、遥信数据处理完回S帧 handleASDU(frame.getAsdu()); recvSeq; writeFrame(ctx, Iec104Frame.buildSFrame(recvSeq)); } } private void sendGeneralCall(ChannelHandlerContext ctx) { // 总召类型100、COT6激活、公共地址按站配置、信息体地址0、QOI0x14 byte[] asdu buildGeneralCallAsdu(); Iec104Frame frame Iec104Frame.buildIFrame(sendSeq, recvSeq, asdu); writeFrame(ctx, frame); } private void writeFrame(ChannelHandlerContext ctx, Iec104Frame frame) { // 这里走编码器统一输出 ctx.writeAndFlush(frame); } }细看这个逻辑我把发送序号和接收序号放在链路Handler里每路连接一份不会串。收到I帧后先业务处理再回S帧S帧携带的接收序号是我期望的下一个序号也就是当前收到序号加1。如果漏了这步从站发送窗口很快会被打满表现就是“连接在数据停”。TESTFR心跳是这里容易被忽略的点。从站端通常要求主站定时发测试帧超过一定时间没收到就判定主站失联并断开链路。我的经验是T3按20秒周期配连续发两次没有TESTFR确认就主动关连接重连比干等TCP超时快得多。4. 编解码器实现粘包拆包与APDU解析4.1 基于长度字段的拆包FrameDecoder核心逻辑粘包拆包是Netty对接104绕不开的一关。TCP是字节流上层发的一帧可能被拆成两半也可能好几帧黏在一起。104自带长度字段L所以用固定头长度字段的方式拆这是最稳的。我写了一个自定义的ByteToMessageDecoder没有直接用LengthFieldBasedFrameDecoder是因为我需要校验同步头0x68和长度范围发现脏数据时能主动丢弃重找。public class Iec104FrameDecoder extends ByteToMessageDecoder { Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, ListObject out) { if (in.readableBytes() 2) { return; } in.markReaderIndex(); byte start in.readByte(); if (start ! 0x68) { // 没有同步头丢掉当前字节继续找 in.resetReaderIndex(); in.skipBytes(1); return; } int len in.readUnsignedByte(); if (len 4 || len 253) { // L非法丢掉0x68重新同步 in.resetReaderIndex(); in.skipBytes(1); return; } if (in.readableBytes() len) { // 半包等下一个TCP段 in.resetReaderIndex(); return; } byte[] frame new byte[2 len]; in.resetReaderIndex(); in.readBytes(frame); out.add(Unpooled.wrappedBuffer(frame)); } }这里最关键的细节是半包处理。in.readableBytes() len时不能直接把已读的头丢出去要resetReaderIndex回到帧头位置等后续字节到齐后再重新走一遍。否则下次decode进来时读取位置已经在半包数据中间永远拼不出完整帧。我见过有人在半包分支里直接return不reset结果每帧数据都被截断联调时全是“接收序号不连续”。脏数据处理也要讲方法。start不等于0x68时只丢1个字节再重新找不能整批丢弃因为0x68可能就藏在下一个字节。L非法时同样只丢同步头把后面内容留给下次解析继续找。这种“逐字节滑窗”看起来慢但104流量不大几路从站完全扛得住。4.2 编码器从Java对象写出68帧编码器反过来把Iec104Frame对象写回ByteBuf。MessageToByteEncoder会自动处理每个出站消息注意writeAndFlush的线程模型这个方法会在IO线程执行所以编码器里不要做耗时操作。public class Iec104FrameEncoder extends MessageToByteEncoderIec104Frame { Override protected void encode(ChannelHandlerContext ctx, Iec104Frame msg, ByteBuf out) { byte[] ctrl msg.getCtrl(); byte[] asdu msg.getAsdu(); out.writeByte(0x68); out.writeByte(4 asdu.length); out.writeBytes(ctrl); out.writeBytes(asdu); } }这个编码器有个默认约定Iec104Frame的ASDU长度不能超过249否则L会溢出变成负数。ASDU本身按规约最大248字节实际业务报文很少到这么大。但如果你在写转发代理把一个长ASDU原样塞进来最好在构建Iec104Frame时先查一下长度超过就分帧。4.3 解析遥测数据类型标识13与品质描述解码器拆出来的只是完整帧真正的业务解析在ASDU上。我用一个静态方法处理最常见的类型13短浮点遥测顺便把品质位拆出来让上层能判断数据是否有效。public final class AsduParser { public static final int TYPE_M_ME_NC 13; // 短浮点遥测 public static final int TYPE_M_SP_NA 1; // 单点遥信 public static ListMeasPoint parseMeas(byte[] asdu) { ListMeasPoint points new ArrayList(); int type asdu[0] 0xFF; int vsq asdu[1] 0xFF; int count vsq 0x7F; boolean continuous (vsq 0x80) 0; // COT在asdu[2..3]公共地址在asdu[4..5]跳过这6个字节到信息体 int pos 6; int addr readUInt24(asdu, pos); pos 3; for (int i 0; i count; i) { if (type TYPE_M_ME_NC) { float value readFloatLE(asdu, pos); int quality asdu[pos 4] 0xFF; points.add(new MeasPoint(addr, value, quality)); pos 5; } else if (type TYPE_M_SP_NA) { int value asdu[pos] 0x01; int quality (asdu[pos] 4) 0x0F; points.add(new MeasPoint(addr, value, quality)); pos 1; } if (continuous) { addr; } else { addr readUInt24(asdu, pos); pos 3; } } return points; } private static int readUInt24(byte[] data, int offset) { return (data[offset] 0xFF) | ((data[offset 1] 0xFF) 8) | ((data[offset 2] 0xFF) 16); } private static float readFloatLE(byte[] data, int offset) { int bits (data[offset] 0xFF) | ((data[offset 1] 0xFF) 8) | ((data[offset 2] 0xFF) 16) | ((data[offset 3] 0xFF) 24); return Float.intBitsToFloat(bits); } }注意readFloatLE这是全网最容易翻车的地方。短浮点按IEEE 754标准在104线路上用小端字节序传输直接用ByteBuf.getFloat默认大端读出来的数会大得离谱或者是个负数。我一开始就是吃这个亏后来对照录波文件才定位到是字节序问题。品质位也不能忽略从站检修时品质字节会变成无效或者溢出你不判断就入库调度大屏上会飘着一条冻住的死数据。5. 量产接入避坑记录超时、序号与双端状态机5.1 现象链路起来了遥测数据却总是“不刷新”原因STARTDT确认后没有发总召或者总召的传送原因写错。104协议里从站默认不会主动把所有数据推给主站主站必须发C_IC_NA类型100的总召唤命令从站才会把全站遥测、遥信以突发方式传上来。只建链不发总召链路是通的数据表永远是空的。解决在收到0x0BSTARTDT确认后立即构造总召ASDU并作为I帧发送。总召的ASDU必须带COT0x06激活、信息体地址全0、召唤限定词QOI0x14总召唤缺一项从站都可能不回。我后来把“发总召”这个动作直接绑进链路状态机的STARTED状态迁移里不依赖外部定时器去触发才彻底解决。还有一层坑总召发出后从站会先回一个总召确认帧COT0x07然后才是大量遥测数据。有些主站实现把“确认帧”当成“数据到了”提前结束等待导致展示层只拿到空表。5.2 现象从站重启后主站一直报接收序号不连续原因从站重启后发送序号清零主站侧还记着重启前的接收序号。双方序号基准不一致I帧校验直接失败链路看起来“活着”但业务数据全被丢弃。解决发现接收序号不连续时不要硬着头皮继续等主动断开连接重建。重建流程里把发送序号、接收序号全部清零重新走STARTDT握手。这是最干净的做法不要尝试在旧序号基础上做偏移补偿104状态机不认。同时把TESTFR心跳重传逻辑补上T3超时发一次TESTFR没确认再发一次连续两次无响应就断链重连这个“心跳包重传”逻辑能帮你从站端异常恢复时自动把链路拉回来。5.3 现象压测时内存飙高、GC频繁原因解码器产出的ByteBuf没有被正确释放。如果直接把ByteBuf传给业务线程池业务还在异步处理时IO线程可能已经把它归还池子数据读到一半变乱码反过来如果不释放内存池会被占满GC压力全上来。解决我统一在channelRead0里先把ByteBuf复制成byte[]再交给业务线程池。这样IO线程的ByteBuf在方法结束后由SimpleChannelInboundHandler自动释放业务线程只管普通byte数组不存在生命周期竞争。代价是多一次拷贝但104数据量远没到需要零拷贝的规模这一份拷贝换来的稳定性非常值。Netty的线程模型也要说清楚channelRead0默认跑在EventLoop线程任何阻塞操作都会卡住整个连接的所有消息。写数据库、调第三方接口必须甩给独立业务线程池。我用一个固定线程池做ASDU入库线程数按核数配遇到从站突发几百个遥测点时不会被IO线程拖死。5.4 现象同时接几十个从站报文串站了原因把公共地址或链路状态做成了静态变量。104主站一般同时服务几十个从站每个从站的公共地址不同、发送窗口不同如果用一个全局Map存接收序号两个从站同时收发时互相覆盖报文解析出来全变成另一站的地址。解决所有链路状态都必须绑在Channel上。我的Iec104LinkHandler用ChannelHandler.Sharable标注的类里不允许有非线程安全的成员变量连接相关的序号、状态全部放在单独的对象里通过AttributeKey附着在Channel上。公共地址从ASDU里读取后再由连接对应的点表配置做映射不能从全局缓存里按“当前站”取。这个坑的排查成本很高因为串站信号在测试环境很稳定一旦上生产接入几十个站就偶发。我用录波文件回放才发现是状态共享导致序号互相污染最后把所有可变状态收敛到连接维度才根治。6. 验证与进阶用模拟从站做一致性回放6.1 先做一个104子站模拟器正式联调前我会先写一个极简模拟从站反向校验自己的主站源码。模拟器起一个ServerSocket收到0x07后回0x0B再收到总召后给对方发固定几帧遥测。这个模拟器不用做全能验证握手、总召、遥测解析就够。public class Iec104Simulator { public static void main(String[] args) throws Exception { EventLoopGroup group new NioEventLoopGroup(1); ServerBootstrap b new ServerBootstrap(); b.group(group) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializerSocketChannel() { Override protected void initChannel(SocketChannel ch) { ch.pipeline().addLast(new SimpleChannelInboundHandlerByteBuf() { Override public void channelActive(ChannelHandlerContext ctx) { // 主动连上来的是主站等STARTDT } Override protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { byte[] data new byte[msg.readableBytes()]; msg.readBytes(data); if ((data[2] 0xFF) 0x07) { // 回STARTDT确认 ctx.writeAndFlush(Unpooled.wrappedBuffer( new byte[]{0x68, 0x04, 0x0B, 0x00, 0x00, 0x00})); } else if ((data[2] 0xFF) 0x02 || (data[2] 0xFF) 0x00) { // 模拟回一帧遥测类型13公共地址1信息体地址1 byte[] frame buildMeasureFrame(); ctx.writeAndFlush(Unpooled.wrappedBuffer(frame)); } } }); } }); b.bind(2404).sync().channel().closeFuture().sync(); } }模拟器跑在你自己的机器上用主站程序去连它。这个反向验证能过滤掉八成的基础问题编解码时序、长度计算、字节序、握手状态。比直接上电科院一致性测试省心太多。6.2 把录波报文回放成测试用例从站厂家通常会提供录波文件或者你在联调现场用抓包工具录一段真实报文。把其中的十六进制字节流存成测试资源在JUnit里直接喂给FrameDecoder断言能拆出预期的帧数、解析出正确的遥测值。Test public void testDecodeRealPacket() { byte[] realFrame hexToBytes( 68 0E 02 00 00 00 64 01 06 00 01 00 00 00 00 00 14); ByteBuf in Unpooled.wrappedBuffer(realFrame); Iec104FrameDecoder decoder new Iec104FrameDecoder(); EmbeddedChannel channel new EmbeddedChannel(decoder); channel.writeInbound(in); ByteBuf out channel.readInbound(); assertArrayEquals(realFrame, ByteBufUtil.getBytes(out)); }这种回放测试的价值在于现场问题发生时你不需要让厂家复现直接把当时的原始报文丢进测试环境改一版测一版。我修过一个非常隐蔽的帧长边界问题就是靠回放定位的。6.3 上线前自查清单检查项自检标准长度校验解码器拒绝L小于4和大于253的帧序号复位断线重连后发送、接收序号归零重走握手总召触发STARTDT确认后自动下发C_IC_NA总召TESTFR心跳T3超时发测试帧连续失败断链重连字节序公共地址、信息体地址、浮点数全部小端解析业务线程池IO线程内无数据库、无阻塞等待状态隔离每连接独立状态禁止全局共享还有一条实战教训上线前一定把日志里的十六进制报文保留一份。104问题诡异起来链路通、数据错、指数级跳变没有原始报文你只能猜。后来我把所有收发的原始帧都按连接归档到日志文件排查时间至少省一半。这套自研方案做完之后我对从站异常重连、序号错乱这类问题的可控程度是直接用黑匣子开源包永远达不到的。希望帮到你。本文还有配套的精品资源点击获取