netty发送socket短连接请求，自定义报文头-526互联

package com.chinaums.japi.util;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.ReferenceCountUtil;

import java.nio.charset.Charset;

public class NettyClient {
    public static void main(String[] args) {
        String req = "<ap><ProductID>IBB</ProductID><ChannelType>ERP</ChannelType><CorpNo/><OpNo/><AuthNo/><ReqSeqNo>TER1694770206768</ReqSeqNo><ReqDate>20230915</ReqDate><ReqTime>173006</ReqTime><Sign/><CCTransCode>ABCDEF</CCTransCode><Amt/><Corp/><Cmp/></ap>";
        try {
            String ret = sendXmlToIct("172.16.16.207", 19999, req);
            System.out.println(ret);
        } catch (InterruptedException e) {
            System.out.println("异常了");
        }
    }

    /**
     * 发送XML报文到ICT
     * 调用该方法时不需要关心报文头，处理器会自动处理
     *
     * @param ip
     * @param port
     * @param xml
     * @return
     * @throws InterruptedException
     */
    public static String sendXmlToIct(String ip, int port, String xml) throws InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 添加处理报文头的 ChannelHandler
                            pipeline.addLast(new MyHeaderReqHandler());
                            pipeline.addLast(new MyHeaderResHandler());
                        }
                    });
            Channel channel = bootstrap.connect(ip, port).sync().channel();
            channel.writeAndFlush(Unpooled.copiedBuffer(xml, Charset.forName("GBK")));
            // 等待消息发送完成
            channel.flush();
            StringBuilder messageBuilder = new StringBuilder();
            channel.pipeline().addLast(new SimpleChannelInboundHandler<ByteBuf>() {
                @Override
                protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
                    messageBuilder.append(msg.toString(Charset.forName("GBK")));
                }

                @Override
                public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
                    super.channelReadComplete(ctx);
                    ctx.channel().close();
                }

                @Override
                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                    super.exceptionCaught(ctx, cause);
                    // 处理异常
                }
            });
            channel.closeFuture().sync();
            return messageBuilder.toString();
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * 发送请求前报文头处理
     * 这个也可以单独弄出去，而不是现在这样的内部类
     * <p>
     * Socket方式中报文结构为“包头+数据包”。
     * 包头固定为7个字节长，第1字节为是否加密标志（0-不加密，1-加密）。
     * 后6个字节是数据包的长度，即将报文长度直接转为字符串存储，长度不足6位则右边用空格补足，
     * 比如：“1234  ”。比如汇兑的长度为1234字节的数据包，其包头为“01234  ”共7位，其中数据包长度包含加密包标志位。
     * 由于加密需要双方约定专门的加密算法，因此一般ERP送的加密标志都为0-不加密。
     */
    public static class MyHeaderReqHandler extends ChannelOutboundHandlerAdapter {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            if (msg instanceof ByteBuf) {
                ByteBuf data = (ByteBuf) msg;
                // 获取数据包长度
                int length = data.readableBytes();
                // 创建一个新的ByteBuf来存储带有报文头的完整数据
                ByteBuf newData = ctx.alloc().buffer(7 + length);
                // 写入加密标志（0-不加密，1-加密）
                newData.writeByte(0); // 这里写入了0，表示不加密
                // 将数据包的长度转换为字符串，并用空格补足6位
                String lengthStr = String.format("%06d", length);
                newData.writeBytes(lengthStr.getBytes());
                // 写入原始数据包
                newData.writeBytes(data);
                // 替换原始消息为带有报文头的新消息
                ReferenceCountUtil.release(msg);
                msg = newData;
            }
            super.write(ctx, msg, promise);
        }
    }

    /**
     * 接收响应后报文头处理
     * 这个也可以单独弄出去，而不是现在这样的内部类
     * <p>
     * Socket方式中报文结构为“包头+数据包”。
     * 包头固定为7个字节长，第1字节为是否加密标志（0-不加密，1-加密）。
     * 后6个字节是数据包的长度，即将报文长度直接转为字符串存储，长度不足6位则右边用空格补足，
     * 比如：“1234  ”。比如汇兑的长度为1234字节的数据包，其包头为“01234  ”共7位，其中数据包长度包含加密包标志位。
     * 由于加密需要双方约定专门的加密算法，因此一般ERP送的加密标志都为0-不加密。
     */
    public static class MyHeaderResHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            if (msg instanceof ByteBuf) {
                ByteBuf data = (ByteBuf) msg;
                // 读取加密标志
                byte encryptFlag = data.readByte();
                // 读取数据包长度
                byte[] lengthBytes = new byte[6];
                data.readBytes(lengthBytes);
                int length = Integer.parseInt(new String(lengthBytes).trim());
                // 读取数据包内容
                ByteBuf packet = data.readBytes(length);
                // 将解析后的数据继续传递给下一个处理器处理
                // 使用retain()方法确保数据不会被提前释放
                ctx.fireChannelRead(packet.retain());
                // 释放原始消息
                ReferenceCountUtil.release(msg);
            }
        }
    }
}