淘客返利机器人高并发 IM 通道：Netty 长连接网关与 MQTT 协议选型对比

大家好，我是 微赚淘客系统3.0 的研发者省赚客！

微赚淘客返利机器人需在用户下单后秒级推送返利到账通知，日均消息量超 500 万条。早期采用 HTTP 轮询方案存在延迟高、设备耗电严重问题。我们重构为基于 长连接的即时消息通道，并在 自研 Netty 网关 与 MQTT 协议中间件 之间进行深度对比与选型，最终实现单机支撑 30 万+ 并发连接、P99 推送延迟 <120ms。

一、核心需求指标

• 支持 50 万+ 在线设备；
• 消息到达率 ≥99.99%；
• 端到端延迟 ≤200ms；
• 低功耗（移动端）；
• 支持 QoS 1（至少一次送达）。

二、方案一：自研 Netty TCP 长连接网关

基于 Netty 实现私有二进制协议，定制心跳、鉴权、消息 ACK 机制。

服务端初始化

package juwatech.cn.im.netty;

public class NettyImServer {
    public void start(int port) {
        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 IdleStateHandler(60, 0, 0)); // 心跳检测
                     p.addLast(new ImMessageDecoder());
                     p.addLast(new ImMessageEncoder());
                     p.addLast(new ImServerHandler()); // 业务处理器
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 1024)
             .childOption(ChannelOption.TCP_NODELAY, true)
             .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

连接管理与消息路由

使用 ConcurrentHashMap 存储 userId → Channel 映射：

package juwatech.cn.im.manager;

@Component
public class ConnectionManager {
    private final Map<Long, Channel> userChannelMap = new ConcurrentHashMap<>();

    public void add(Long userId, Channel channel) {
        userChannelMap.put(userId, channel);
    }

    public void remove(Long userId) {
        userChannelMap.remove(userId);
    }

    public void sendToUser(Long userId, ImMessage msg) {
        Channel ch = userChannelMap.get(userId);
        if (ch != null && ch.isActive()) {
            ch.writeAndFlush(msg);
        } else {
            // 离线消息存 DB，上线后拉取
            offlineMessageService.save(userId, msg);
        }
    }
}

协议设计（简化版）

// 消息头：4字节长度 + 1字节类型 + 8字节 traceId
// 消息体：JSON payload
public class ImMessage {
    private byte type;      // 1: text, 2: rebate_notify
    private String payload; // {"tradeId":"T123","amount":2.99}
    private String traceId;
}

优势：完全可控、低延迟、无第三方依赖。
劣势：需自行实现重连、QoS、集群广播等能力。

三、方案二：基于 EMQX 的 MQTT 协议

采用开源 MQTT Broker EMQX，客户端使用 Paho SDK。

客户端连接（Android 示例）

MqttConnectOptions options = new MqttConnectOptions();
options.setUserName("user_10086");
options.setPassword("token_xxx".toCharArray());
options.setCleanSession(false); // 保留离线消息
options.setKeepAliveInterval(60);

MqttClient client = new MqttClient("tcp://mqtt.juwatech.cn:1883", "client_10086");
client.connect(options);

// 订阅个人通知主题
client.subscribe("rebate/user/10086", 1); // QoS=1

服务端推送

package juwatech.cn.im.mqtt;

@Service
public class MqttPushService {

    @Autowired
    private MqttClient mqttClient; // 共享客户端

    public void pushRebateNotify(Long userId, RebateNotify notify) {
        String topic = "rebate/user/" + userId;
        String payload = JsonUtils.toJson(notify);
        MqttMessage message = new MqttMessage(payload.getBytes());
        message.setQos(1);
        message.setRetained(false);
        mqttClient.publish(topic, message);
    }
}

EMQX 支持集群、规则引擎（可写入 Kafka）、Dashboard 监控。

优势：标准协议、生态完善、自动处理重传与会话保持。
劣势：引入外部组件、协议开销略高（UTF-8 主题、固定头）。

四、性能压测对比

在 16C32G 服务器上部署单节点：

指标	Netty 自研	EMQX (MQTT)
最大连接数	320,000	280,000
消息吞吐 (msg/s)	85,000	72,000
P99 延迟	98ms	115ms
内存占用	4.2GB	5.8GB
开发成本	高（需实现完整 IM 逻辑）	低（聚焦业务）

五、最终选型与混合架构

• 核心通知（返利到账、提现成功）：采用 Netty 自研网关，保障最低延迟；
• 营销消息（优惠券、活动）：走 MQTT 通道，利用其广播与离线队列能力；
• 连接桥接：通过 userId 统一路由，若 Netty 连接断开，自动降级至 MQTT。

连接状态统一管理：

public enum ConnectionType { NETTY, MQTT }

public class UnifiedPushService {
    public void push(Long userId, Message msg) {
        ConnectionType type = connectionRegistry.getType(userId);
        if (type == ConnectionType.NETTY) {
            nettyPushService.send(userId, msg);
        } else {
            mqttPushService.push(userId, msg);
        }
    }
}

该混合架构在双十一大促期间稳定运行，消息到达率 99.997%，平均推送延迟 86ms。

本文著作权归 微赚淘客系统3.0 研发团队，转载请注明出处！