基于Netty与Protobuf实现微信个人号iPad协议长连接网关的Java实战

在模拟微信个人号iPad客户端的私有协议对接场景中，需建立稳定的长连接以接收消息推送、心跳保活及指令响应。由于微信iPad协议采用自定义二进制格式（头部+Protobuf载荷），使用 Netty 作为网络层、Protobuf 作为序列化方案是高效可靠的选择。本文展示一个可运行的长连接网关核心实现。

1. 协议帧结构解析

微信iPad协议帧通常包含：

• 4字节 Magic（固定值 0x0A 0x00 0x00 0x00）
• 4字节 Length（后续数据长度）
• N字节 Protobuf 序列化数据

定义解码器：

public class WechatMessageDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        if (in.readableBytes() < 8) return;

        in.markReaderIndex();
        int magic = in.readInt();
        if (magic != 0x0A000000) {
            in.resetReaderIndex();
            ctx.close(); // 非法协议
            return;
        }

        int length = in.readInt();
        if (in.readableBytes() < length) {
            in.resetReaderIndex();
            return;
        }

        byte[] payload = new byte[length];
        in.readBytes(payload);

        try {
            WechatCommand command = WechatCommand.parseFrom(payload);
            out.add(command);
        } catch (InvalidProtocolBufferException e) {
            ctx.close();
        }
    }
}

其中 WechatCommand 由 .proto 文件生成：

syntax = "proto3";
message WechatCommand {
  int32 cmd_id = 1;
  bytes data = 2;
}

2. Netty 服务端启动

配置 EventLoopGroup 与 ChannelPipeline：

public class WechatGatewayServer {

    public void start(int port) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 4, 4, -8, 0));
                     p.addLast(new WechatMessageDecoder());
                     p.addLast(new WechatMessageEncoder());
                     p.addLast(new WechatMessageHandler());
                 }
             });

            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

注意：此处先用 LengthFieldBasedFrameDecoder 处理长度字段，再交由自定义 WechatMessageDecoder 解析 Protobuf。

3. 消息处理器与业务路由

实现 ChannelInboundHandlerAdapter 处理命令：

public class WechatMessageHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if (msg instanceof WechatCommand) {
            WechatCommand cmd = (WechatCommand) msg;
            int cmdId = cmd.getCmdId();

            switch (cmdId) {
                case 1001: // 心跳
                    sendHeartbeatResponse(ctx);
                    break;
                case 2001: // 新消息
                    handleNewMessage(ctx, cmd.getData());
                    break;
                default:
                    // 未知命令
            }
        }
    }

    private void handleNewMessage(ChannelHandlerContext ctx, ByteString data) {
        try {
            MessageContent content = MessageContent.parseFrom(data);
            String fromUser = content.getFromUser();
            String text = content.getText();

            // 转发至业务服务
            wlkankan.cn.service.MessageRouter.route(fromUser, text);

            // 回复ACK
            WechatCommand ack = WechatCommand.newBuilder()
                .setCmdId(2002)
                .setData(ByteString.copyFromUtf8("OK"))
                .build();
            ctx.writeAndFlush(ack);
        } catch (InvalidProtocolBufferException e) {
            ctx.close();
        }
    }

    private void sendHeartbeatResponse(ChannelHandlerContext ctx) {
        WechatCommand resp = WechatCommand.newBuilder()
            .setCmdId(1002)
            .build();
        ctx.writeAndFlush(resp);
    }
}

4. 消息编码器实现

将 WechatCommand 序列化为协议帧：

public class WechatMessageEncoder extends MessageToByteEncoder<WechatCommand> {

    @Override
    protected void encode(ChannelHandlerContext ctx, WechatCommand msg, ByteBuf out) {
        byte[] payload = msg.toByteArray();
        out.writeInt(0x0A000000); // Magic
        out.writeInt(payload.length); // Length
        out.writeBytes(payload);      // Payload
    }
}

5. 连接管理与上下文绑定

为每个设备绑定唯一标识，便于定向推送：

@Component
public class DeviceConnectionManager {

    private final Map<String, Channel> deviceChannelMap = new ConcurrentHashMap<>();

    public void bindDevice(String deviceId, Channel channel) {
        deviceChannelMap.put(deviceId, channel);
        channel.attr(AttributeKey.valueOf("deviceId")).set(deviceId);
    }

    public void unbindDevice(Channel channel) {
        String deviceId = channel.attr(AttributeKey.valueOf("deviceId")).get();
        if (deviceId != null) {
            deviceChannelMap.remove(deviceId);
        }
    }

    public void sendMessageToDevice(String deviceId, WechatCommand cmd) {
        Channel channel = deviceChannelMap.get(deviceId);
        if (channel != null && channel.isActive()) {
            channel.writeAndFlush(cmd);
        } else {
            // 离线处理：存入队列或丢弃
            wlkankan.cn.queue.OfflineMessageQueue.enqueue(deviceId, cmd);
        }
    }
}

在 WechatMessageHandler 的 channelActive 和 channelInactive 中调用绑定/解绑：

@Override
public void channelActive(ChannelHandlerContext ctx) {
    // 登录认证逻辑（略）
    String deviceId = authenticate(ctx);
    wlkankan.cn.connection.DeviceConnectionManager.bindDevice(deviceId, ctx.channel());
}

@Override
public void channelInactive(ChannelHandlerContext ctx) {
    wlkankan.cn.connection.DeviceConnectionManager.unbindDevice(ctx.channel());
}

6. 心跳与空闲检测

防止僵尸连接占用资源：

// 在 ChannelInitializer 中添加
p.addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS));
p.addLast(new HeartbeatHandler());

// 心跳处理器
public class HeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            ctx.close(); // 60秒无读事件，断开
        }
    }
}

通过上述组件，可构建一个高并发、低延迟的微信个人号iPad协议长连接网关，支撑消息收发、设备管理与实时指令下发。