★ FEATURED ARTICLE 1. 为什么要把 Kafka 和 Netty 放进同一个 Spring Boot 工程很多刚接触消息中间件和网络编程的朋友第一次看到Spring Boot 整合 Kafka 和 Netty这个组合时脑子里冒出的第一个问题往往是这俩东西八竿子打不着为什么要塞进一个工程里我当初也是这么想的直到真正做过几个物联网数据采集和实时推送类的项目之后才明白这个组合其实解决的是一个非常具体的场景问题。先说结论Kafka 负责削峰、解耦、持久化Netty 负责高并发长连接接入和实时下发。这两者的定位完全不同但放在一起刚好能拼出一条完整的链路。举个我实际遇到过的场景——假设你有一批设备或者客户端需要保持长连接持续上报数据同时服务端还要能主动把消息推给指定的连接。如果只用 HTTP 轮询连接开销大、实时性差如果只用 Netty 自己扛消息堆积、服务重启丢数据、多实例之间没法共享消费进度这些问题会一个接一个冒出来。这时候把 Kafka 拉进来就顺理成章了。Netty 收到客户端上报的数据后不直接做业务处理而是先丢进 Kafka 的某个 Topic下游的业务服务作为消费者去订阅这个 Topic慢慢处理。反过来业务侧要推送消息给某个客户端时也先写进 Kafka再由持有该客户端连接的 Netty 节点消费并下发。这样一来接入层和业务层彻底解耦Netty 节点可以水平扩容业务处理速度也不再受限于网络 IO。我见过不少团队一开始图省事Netty 收到消息直接在 ChannelHandler 里写数据库、调接口结果高峰期线程池打满连接被拖垮最后不得不返工重构成Netty MQ的架构。所以这个简单实例虽然叫简单但它背后的架构思路一点都不简单值得认真拆一拆。这篇文章我会从零开始把环境准备、Kafka 生产消费的封装、Netty 服务端的搭建、两者的桥接、以及实际跑起来之后容易踩的坑全部讲清楚。代码部分我会给出可以直接复制的片段配置部分我会解释每个参数为什么这么设。适合有一定 Spring Boot 基础、想入门消息队列和网络编程整合的开发者也适合正在做实时数据链路选型的技术负责人参考。2. 动手前的环境与依赖梳理2.1 版本选择这件事别随便拍脑袋整合类项目最容易翻车的地方就是版本。Spring Boot、Spring Kafka、Kafka 客户端、Netty 这四者之间存在兼容性关系选错了轻则启动报错重则运行时出现诡异的序列化问题。我个人的经验是Spring Boot 用 2.7.x 或 3.x 的稳定版Spring Kafka 跟着 Spring Boot 的 BOM 走不要手动指定版本这样能避免绝大部分依赖冲突。下面是我实测比较稳的一套组合供参考组件版本说明Spring Boot2.7.18生态成熟资料多适合入门Spring Kafka随 BOM由 spring-boot-starter-parent 统一管理Kafka 服务端3.x本地用 Docker 起一个单节点即可Netty4.1.x由 spring-boot-starter 间接引入或手动指定如果你用的是 Spring Boot 3.x注意它要求 JDK 17 起步而且部分 Jakarta 包名发生了变化老代码直接搬过去会编译不过。入门阶段我建议先用 2.7.x 把链路跑通理解清楚之后再升级。2.2 依赖引入与本地 Kafka 的启动Maven 依赖其实很干净核心就三个dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency dependency groupIdorg.springframework.kafka/groupId artifactIdspring-kafka/artifactId /dependency dependency groupIdio.netty/groupId artifactIdnetty-all/artifactId version4.1.100.Final/version /dependency本地起 Kafka 我最推荐用 Docker Compose一条命令搞定不用去折腾那些脚本version: 3 services: zookeeper: image: zookeeper:3.8 ports: - 2181:2181 kafka: image: bitnami/kafka:3.6 ports: - 9092:9092 environment: KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_CFG_LISTENERS: PLAINTEXT://:9092 KAFKA_CFG_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092 depends_on: - zookeeper提示新版 Kafka 已经可以脱离 Zookeeper 用 KRaft 模式运行但入门阶段用 Zookeeper 模式资料更多、报错更好查等熟悉了再换 KRaft 不迟。启动之后用kafka-topics.sh建两个 Topic一个用于上行数据设备到服务端一个用于下行数据服务端到设备。Topic 的分区数建议至少设成和你的 Netty 节点数一致这样消费才能并行起来。分区数一旦定下来后期只能增不能减所以宁可一开始多设几个。2.3 配置文件里那几个必须理解的参数application.yml里 Kafka 的配置看着简单但每一项都影响行为spring: kafka: bootstrap-servers: localhost:9092 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer acks: all retries: 3 consumer: group-id: netty-bridge-group auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer enable-auto-commit: false这里有几个点我要重点说一下。acks: all表示所有同步副本都确认才算写入成功可靠性最高但延迟略大做设备数据采集这种不能丢的场景必须这么设。enable-auto-commit: false是很多新手会忽略的——自动提交 offset 会在消息还没处理完就提交一旦处理过程中服务挂了这条消息就永远丢了。手动提交虽然代码多几行但能保证至少处理一次。auto-offset-reset: earliest决定了消费者组第一次启动、或者 offset 失效时从哪里开始读。设成 earliest 会从头读适合调试生产环境如果不想重复消费历史数据可以设成 latest。这个参数我踩过坑有一次测试环境误设成 latest结果重启后一批消息没消费到排查了半天才发现是这个配置的锅。3. Kafka 生产与消费的封装思路3.1 别把 KafkaTemplate 到处乱塞很多示例代码喜欢在业务类里直接Autowired KafkaTemplate然后到处send()。小项目无所谓但一旦要统一加日志、加埋点、加异常兜底这种写法就会让你改到崩溃。我的做法是封装一个 MessageProducer 组件把发送逻辑收口Component public class MessageProducer { private final KafkaTemplateString, String kafkaTemplate; public MessageProducer(KafkaTemplateString, String kafkaTemplate) { this.kafkaTemplate kafkaTemplate; } public void send(String topic, String key, String payload) { kafkaTemplate.send(topic, key, payload) .addCallback( result - log.info(发送成功 topic{} offset{}, topic, result.getRecordMetadata().offset()), ex - log.error(发送失败 topic{} payload{}, topic, payload, ex) ); } }用 key 来指定分区是个实用技巧。Kafka 默认按 key 的 hash 决定分区如果你把设备 ID 作为 key那么同一台设备的消息就会落到同一个分区消费时天然有序。这一点在设备指令下发场景里特别重要——指令顺序错了设备行为就乱了。3.2 消费者用 KafkaListener 还是自己写容器Spring Kafka 提供了KafkaListener注解几行代码就能起一个消费者非常方便。但如果你需要动态控制消费开关、或者消费逻辑要跟 Netty 的 Channel 状态联动注解方式就会显得不够灵活。我的建议是普通业务消费用注解跟 Netty 桥接的那部分用编程式容器。注解方式的写法KafkaListener(topics device-uplink, groupId business-group) public void onUplink(String message, Acknowledgment ack) { try { // 业务处理 process(message); ack.acknowledge(); } catch (Exception e) { log.error(处理失败不提交 offset等待重试, e); } }注意这里配合enable-auto-commit: false和AckMode.MANUAL只有业务真正处理成功才提交。失败就不提交Kafka 会重新投递。当然这要求你的业务逻辑是幂等的否则重复消费会出问题——这是消息队列的通用课题后面我会专门讲。3.3 序列化选 String 还是 JSON入门阶段我强烈建议统一用 String 序列化业务对象自己用 Jackson 转 JSON。原因有两个一是排查问题时你在 Kafka 命令行工具里能直接看到明文消息非常直观二是避免了自定义 Serializer 带来的类加载和版本兼容问题。等链路稳定了再考虑上 Avro、Protobuf 这类带 schema 的方案做性能优化。我见过有团队一上来就用自定义序列化器结果消费端反序列化失败日志里全是乱码排查成本极高。先用 String 把链路跑通这个顺序不能反。4. Netty 服务端的搭建与关键配置4.1 线程模型bossGroup 和 workerGroup 到底怎么分Netty 的线程模型是它的核心优势但也是新手最容易配错的地方。简单说bossGroup负责接收连接workerGroup负责处理已建立连接的 IO 读写。对于绝大多数场景bossGroup 设 1 个线程就够了因为它只做 accept压力很小workerGroup 默认是 CPU 核数乘 2这个值在 IO 密集型场景下通常够用。EventLoopGroup bossGroup new NioEventLoopGroup(1); EventLoopGroup workerGroup new NioEventLoopGroup(); try { ServerBootstrap bootstrap new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childOption(ChannelOption.SO_KEEPALIVE, true) .childOption(ChannelOption.TCP_NODELAY, true) .childHandler(new ChannelInitializerSocketChannel() { Override protected void initChannel(SocketChannel ch) { ch.pipeline() .addLast(new IdleStateHandler(60, 0, 0, TimeUnit.SECONDS)) .addLast(new StringDecoder()) .addLast(new StringEncoder()) .addLast(new DeviceHandler()); } }); ChannelFuture future bootstrap.bind(8888).sync(); future.channel().closeFuture().sync(); } finally { bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); }SO_BACKLOG是等待 accept 的连接队列长度高并发接入时设大一点能减少连接被拒。TCP_NODELAY关闭 Nagle 算法对于小包频繁发送的场景能显著降低延迟——设备指令通常都很短这个参数很关键。4.2 心跳与连接管理IdleStateHandler 的正确用法长连接最怕的就是假死——TCP 连接还在但对端其实已经挂了。IdleStateHandler就是解决这个问题的。上面代码里new IdleStateHandler(60, 0, 0)表示 60 秒没读到数据就触发读空闲事件你可以在 Handler 里捕获这个事件主动关闭连接。Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) { if (evt instanceof IdleStateEvent) { IdleStateEvent event (IdleStateEvent) evt; if (event.state() IdleState.READER_IDLE) { log.warn(连接空闲超时关闭 channel{}, ctx.channel().id()); ctx.close(); } } }这里有个经验心跳超时时间要跟客户端约定好服务端设的值要比客户端心跳间隔大 2 到 3 倍。比如客户端 20 秒发一次心跳服务端设 60 秒超时比较稳妥能容忍一两次网络抖动。设太紧会误杀正常连接设太松则假死连接清理不及时。4.3 Channel 的集中管理推送的前提要做服务端主动推送你必须知道某个用户对应哪个 Channel。所以需要一个全局的 Channel 管理器通常用ConcurrentHashMap存userId - Channel的映射Component public class ChannelManager { private final MapString, Channel channels new ConcurrentHashMap(); public void bind(String userId, Channel channel) { channels.put(userId, channel); } public void unbind(String userId) { channels.remove(userId); } public Channel get(String userId) { return channels.get(userId); } }注意Channel 关闭时一定要记得从 map 里移除否则会内存泄漏。我一般会在channelInactive回调里做清理同时把 userId 和 channel 的绑定关系也解掉。这个管理器是单机内存级的如果你的 Netty 是多节点部署那用户连在哪个节点上这个问题就需要额外的路由信息。这也是为什么下行消息要走 Kafka——业务侧只管把消息发到 Kafka各个 Netty 节点都订阅收到后检查自己有没有这个用户的 Channel有就推送没有就忽略。这种广播 本地过滤的模式虽然有点浪费但实现简单、可靠性高是入门阶段最实用的方案。5. 把 Kafka 和 Netty 真正接起来5.1 上行链路Netty 收到数据后写 Kafka这是整条链路里最核心的一环。Netty 的 Handler 收到客户端消息后不要做任何耗时业务直接丢给 Kafkapublic class DeviceHandler extends SimpleChannelInboundHandlerString { private final MessageProducer producer; public DeviceHandler(MessageProducer producer) { this.producer producer; } Override protected void channelRead0(ChannelHandlerContext ctx, String msg) { String deviceId parseDeviceId(msg); producer.send(device-uplink, deviceId, msg); } }这里有个细节要注意Handler 是每个连接一个实例但 MessageProducer 应该是全局单例。所以创建 Handler 的时候要把 producer 传进去而不是在 Handler 里 new。我见过有人在 Handler 里直接new KafkaProducer()结果每个连接都创建一个生产者连接数一多直接把 broker 打爆。另外channelRead0里如果做同步发送.get()会阻塞 Netty 的 IO 线程这是大忌。上面用的addCallback是异步的不会阻塞。如果确实需要确认发送结果也应该切到业务线程池去处理绝不能占用 EventLoop。5.2 下行链路消费 Kafka 消息并推送到指定 Channel下行就是一个 Kafka 消费者收到消息后查 ChannelManager找到对应连接就写回去KafkaListener(topics device-downlink, groupId netty-bridge-group) public void onDownlink(String message) { DownlinkMsg msg JSON.parseObject(message, DownlinkMsg.class); Channel channel channelManager.get(msg.getUserId()); if (channel ! null channel.isActive()) { channel.writeAndFlush(msg.getContent() \n); } else { log.warn(用户 {} 不在线消息丢弃或转离线存储, msg.getUserId()); } }writeAndFlush是异步的返回 ChannelFuture正常情况下不用等。但如果你的推送量非常大建议加个判断channel.isWritable()如果写缓冲区满了就先别写否则会堆积在内存里导致 OOM。这是高并发推送场景下的一个重要保护措施。5.3 消息格式约定别让两端各说各话上行下行都涉及消息格式一定要提前约定好。我推荐用简单的 JSON字段固定{ type: command, userId: u1001, deviceId: d2001, content: REBOOT, timestamp: 1700000000000 }type用来区分消息种类userId用于路由timestamp用于排查时序问题。别小看这个约定我做过一个项目因为两端对字段名理解不一致一边用 userId 一边用 uid联调时白白浪费了一整天。6. 实测中那些文档不会告诉你的坑6.1 消费重复与幂等这不是 Kafka 的锅前面提到手动提交 offset 保证至少一次代价就是可能重复消费。比如消息处理完了、offset 还没提交这时候服务重启这条消息会被重新投递。这不是 bug是消息队列的语义决定的。解决办法是业务侧做幂等给每条消息带一个全局唯一的 messageId处理前先查一下这个 ID 有没有处理过处理过就直接跳过。我一般会在 Redis 里存一个processed:{messageId}的标记设置合理的过期时间比如 24 小时。这样既能防重复又不会无限占用内存。注意这个查重 处理 标记的过程最好用分布式锁或者原子操作保护否则并发下还是可能重复。6.2 Netty 的 ByteBuf 内存泄漏Netty 用引用计数管理 ByteBuf如果用了SimpleChannelInboundHandler它会自动释放消息你不用管。但如果你继承的是ChannelInboundHandlerAdapter就必须自己调用ReferenceCountUtil.release(msg)否则会内存泄漏。这个坑非常隐蔽泄漏初期看不出来跑几天之后堆外内存爆掉日志里会出现LEAK: ByteBuf.release() was not called的警告。我的建议是入门阶段一律用SimpleChannelInboundHandler省心。等对引用计数机制理解透了再根据性能需要选择更底层的写法。6.3 优雅关闭别让服务重启丢消息服务关闭时如果直接 kill 进程正在处理的消息可能丢Kafka 的 offset 也可能没提交。正确做法是注册一个关闭钩子先停止接收新连接再等正在处理的消息完成最后关闭 Kafka 消费者和 Netty 线程组Runtime.getRuntime().addShutdownHook(new Thread(() - { log.info(开始优雅关闭); bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); // 等待 Kafka 消费者处理完当前批次 }));Spring Boot 本身有优雅关闭机制配合server.shutdown: graceful和合理的超时时间能大幅降低重启导致的数据问题。这个配置在容器化部署、频繁发布的环境里尤其重要。6.4 分区数与消费并行度的匹配Kafka 的消费并行度上限就是分区数——一个分区只能被同一个消费者组里的一个消费者消费。如果你 Topic 只设了 1 个分区那不管你起多少个消费者实例实际只有一个在干活。我见过有人抱怨消费太慢查了半天发现是 Topic 分区数设成了 1。经验值分区数 预期消费者实例数 × 每个实例的消费线程数再留一点余量。比如你计划部署 4 个 Netty 节点每个节点 2 个消费线程那分区数设 8 到 12 比较合适。分区多了会增加 broker 的元数据开销但现代 Kafka 扛几千个分区没问题不用太保守。7. 从能跑到能扛的几点优化方向链路跑通只是第一步真正上线还要考虑性能和稳定性。这里分享几个我实际用过的优化点。批量发送。Kafka 生产者有个linger.ms参数默认是 0表示消息立即发送。设成 5 到 10 毫秒生产者会攒一批再发吞吐量能提升好几倍代价是延迟增加几毫秒。对于设备数据采集这种吞吐优先的场景这个取舍很划算。配合batch.size调大一点效果更明显。Netty 的写缓冲水位。通过WriteBufferWaterMark设置高低水位当待写数据超过高水位时isWritable()返回 false你就可以暂停推送等降到低水位再继续。这是防止慢客户端拖垮服务端的标准做法。监控埋点。至少要有这几个指标Kafka 生产/消费的速率和延迟、Netty 的活跃连接数、Channel 的读写字节数、消费积压量lag。消费积压是最重要的告警指标一旦持续增长说明消费能力跟不上生产速度得赶紧扩容或者优化消费逻辑。压测。上线前一定要压。用工具模拟几千个长连接同时上报观察 Netty 的内存、CPU以及 Kafka 的 lag 变化。我踩过的坑是单机测试几百连接一切正常一上生产几千连接就出现连接超时最后发现是操作系统的文件描述符限制没调。ulimit -n这个值默认往往只有 1024长连接服务必须调大。8. 一些个人体会这套Spring Boot Kafka Netty的组合我前后在好几个项目里用过从最初的照猫画虎到后来的得心应手中间踩的坑基本都写在上面的章节里了。如果让我给刚上手的朋友一句建议那就是先把最简单的链路跑通再谈优化。不要一上来就想着集群、幂等、监控全套上齐那样很容易在环境问题上卡住还没体会到架构的好处就先被劝退了。具体来说我的推荐路径是单机单分区跑通上行链路确认消息能进 Kafka再加下行链路确认能推回客户端然后才是多分区、多实例、幂等、监控。每一步都验证清楚再往下走出问题时排查范围小定位快。另外Kafka 和 Netty 各自都是深坑极多的技术这个实例只是把它们串起来的骨架。真正要做好Kafka 的副本机制、ISR、再平衡Netty 的零拷贝、内存池、编解码每一项都值得单独深入研究。但骨架先立起来后面往里填肉就顺了。希望这篇内容能帮你少走点弯路把这条实时数据链路稳稳当当地跑起来。
阅读完成 · 觉得有帮助？
