咱就是说，现在搞后端开发，谁还没听说过 Kafka 嘞？但很多刚开始接触的朋友，对 Kafka 到底能干啥、该用在哪，其实还是一脸懵的——"不就是个消息队列嘛，我用 Redis 发个消息不也行？"

嘿嘿，今天咱就来聊聊 Kafka 那些常见的使用场景，从入门到实战，把 Kafka 这张王牌彻底整明白！

一、异步解耦——让微服务学会"各玩各的"

想象一下这个场景：用户注册后，我们需要做三件事——发短信通知、送新用户优惠券、把数据同步到数据仓库。

如果咱写成一个同步链路：

用户注册 → 发短信 → 送优惠券 → 同步数仓 → 返回"注册成功"

这用户得等多久嘞？万一发短信的服务挂了，整个注册流程都得跟着炸，这谁能忍啊！

异步解耦的思路就很简单粗暴了：注册服务干完自己的活（写入数据库），往 Kafka 丢一条消息就完事了，剩下的爱谁谁。

// 注册服务：干完活直接发消息，潇洒走人
@Service
public class RegisterService {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void register(User user) {
        // 1. 写入数据库，这个是我的核心职责
        userMapper.insert(user);
        
        // 2. 丢一条消息出去，剩下的我不管了 😎
        kafkaTemplate.send("user-register-topic", JSON.toJSONString(user));
    }
}

然后短信服务、优惠券服务、数仓服务各自订阅这个 topic，各玩各的，互不影响：

// 短信服务：我只关心发短信
@KafkaListener(topics = "user-register-topic")
public void sendSMS(String message) {
    User user = JSON.parseObject(message, User.class);
    smsService.send(user.getPhone(), "欢迎注册！");
}

// 优惠券服务：我只关心发券
@KafkaListener(topics = "user-register-topic")
public void sendCoupon(String message) {
    User user = JSON.parseObject(message, User.class);
    couponService.grant(user.getId(), "新人券");
}

这样一来，注册服务根本不需要知道下游有哪些消费者，下游新增一个消费者也完全不用改动注册服务的代码。这就是"解耦"的魅力——各个服务高内聚低耦合，今天我们想把短信服务换成推送通知，直接换个消费者就行，上游连一行代码都不带改的。

> 小提示：这里不同的消费者最好用不同的 `groupId`，这样每条消息每个组都能收到。如果用了同一个 groupId，那一条消息只会被组内一个消费者拿到，其他服务就收不到了嗷！

二、削峰填谷——别让流量把数据库"冲烂了"

每年的双十一、618，秒杀活动的流量那是相当吓人的。假设咱的数据库每秒能抗 2000 QPS，结果大促的时候瞬间来了 20000 QPS，数据库直接就"去二次元了"（炸了）。

这时候怎么办呢？不能让所有请求都直接打到数据库上。

Kafka 削峰填谷的逻辑是：把请求先全部扔进 Kafka，然后消费者按自己能消化的速度慢慢消费。流量还是那个流量，但消费速度从"一秒钟灌进去"变成"慢慢喝"，数据库压力就稳住了。

// 秒杀接口：请求来了我照单全收，但不是直接操作数据库
@PostMapping("/seckill")
public Result seckill(Long productId, Long userId) {
    OrderMessage msg = new OrderMessage(productId, userId, System.currentTimeMillis());
    // 直接扔进 Kafka，接口轻松应对高并发
    kafkaTemplate.send("seckill-order-topic", JSON.toJSONString(msg));
    return Result.ok("排队中，请稍后查看结果");  // 异步返回，不阻塞用户
}

// 消费者：慢慢处理订单，数据库完全扛得住
@KafkaListener(topics = "seckill-order-topic", concurrency = "3")
public void processOrder(String message) {
    OrderMessage msg = JSON.parseObject(message, OrderMessage.class);
    orderService.createOrder(msg.getProductId(), msg.getUserId());
}

你看看，本来一秒来两万的请求，经过 Kafka 一缓冲，消费者每秒处理一千条，数据库稳如老狗。

用 Kafka 做削峰填谷，和直接限流的区别在于：限流是直接拒绝多余请求（用户看到"系统繁忙"），削峰填谷是所有请求都能受理，就是处理慢点，体验好太多了。

三、延迟消息——该来的总会来，但可以晚点来

有些场景是这样的：用户下单后 30 分钟内如果不支付，咱要自动取消订单。

这个 30 分钟的延迟，咋搞嘞？

传统的做法可能是弄个定时任务，每隔几分钟扫一遍数据库——这方案也不是不行，但订单表动不动就上千万行，你扫一个我看看？扫一次数据库就哭一次，而且定时任务的精度也跟不上。

其实 Kafka 没有原生的延迟消息功能（RocketMQ 有），但我们可以绕个弯子——多级时间轮的思想，把消息按延迟时间丢到不同的 topic 里，用一个调度器来转发：

// 下单后发延迟消息到 30min 级别
public void orderCreated(Order order) {
    DelayMessage msg = new DelayMessage();
    msg.setOrderId(order.getId());
    msg.setExpireTime(System.currentTimeMillis() + 30 * 60 * 1000);
    // 先写到一个外部存储，比如 Redis，存下消息体
    redisTemplate.opsForValue().set("delay:" + order.getId(), JSON.toJSONString(msg));
    // 同时写到一个有定时触发的 topic
    kafkaTemplate.send("delay-order-topic", JSON.toJSONString(msg));
}

// 延迟调度器：轮询检查哪些消息该投递了
@Scheduled(fixedDelay = 5000)
public void delayDispatcher() {
    // 从 Redis 扫出到期的消息
    Set<String> keys = redisTemplate.keys("delay:*");
    for (String key : keys) {
        DelayMessage msg = JSON.parseObject(redisTemplate.opsForValue().get(key), DelayMessage.class);
        if (msg.getExpireTime() <= System.currentTimeMillis()) {
            // 时间到了，投到目标 topic
            kafkaTemplate.send("order-cancel-topic", JSON.toJSONString(msg));
            redisTemplate.delete(key);
        }
    }
}

不过说实话，如果是 Kafaka 做延迟消息，咱不建议搞得太复杂。Kafka 其实可以通过 时间戳索引 来实现类似效果——在 `Consumer` 消费时判断消息的延迟时间是否到了，没到就 `pause()` 一下分区，等时间差不多了再 `resume()`。

如果延迟消息是你们的刚需，还是直接上 RocketMQ 更省心，Kafka 在这方面确实不是原生的优势。毕竟人家 RocketMQ 原生就带了 18 个延迟等级，开箱即用，这个咱得承认嘞（狗头）。

四、事务消息——要么全做，要么全不做

事务消息听起来很高大上，其实说白了就一句话：保证"发消息"和"执行本地业务"这两个操作要么都成功，要么都失败。

典型的场景就是下单扣库存：订单创建成功 + 发消息通知扣库存，这两个操作不能出现"订单已经创建了，但扣库存的消息没发出去"的情况。

传统的做法是在数据库里建一张"本地消息表"，用同一个本地事务包裹：

@Transactional
public void createOrder(Order order) {
    // 1. 写订单表
    orderMapper.insert(order);
    // 2. 写本地消息表（和订单在同一个数据库事务里）
    MessageLog log = new MessageLog();
    log.setTopic("order-created");
    log.setPayload(JSON.toJSONString(order));
    log.setStatus("PENDING");
    messageLogMapper.insert(log);
    // 只要事务提交成功，这两条记录一定都存在
}

然后另开一个定时任务，把消息表中 `PENDING` 的消息发到 Kafka，成功发送后更新状态为 `SENT`：

@Scheduled(fixedDelay = 3000)
public void sendPendingMessages() {
    List<MessageLog> pending = messageLogMapper.findByStatus("PENDING");
    for (MessageLog log : pending) {
        try {
            kafkaTemplate.send(log.getTopic(), log.getPayload()).get();
            log.setStatus("SENT");
            messageLogMapper.update(log);
        } catch (Exception e) {
            // 失败了没事，下次定时任务再捞起来重试
            log.warn("消息发送失败，等待重试: {}", log.getId());
        }
    }
}

这套方案在很多公司里都在用，稳得很。

那 Kafka 原生的"事务"是啥呢？ 其实 Kafka 的事务主要是保证生产者幂等 + 跨分区原子写入，跟我们常说的"分布式事务"不是一回事。Kafka 的事务更像是在说："放心吧，你发的这坨消息要么全部被 consumer 看见，要么一条都看不见，不会出现只看见一半的情况。"

如果真要做分布式事务，Kafka 事务 + 本地消息表一起上会比较稳：

// Kafka 事务消息示例（生产者端）
producer.initTransactions();
try {
    producer.beginTransaction();
    // 发多条消息
    producer.send(new ProducerRecord<>("topic-A", "msg1"));
    producer.send(new ProducerRecord<>("topic-B", "msg2"));
    // 本地业务也一起提交
    localService.doBusiness();
    producer.commitTransaction();  // 一起成功
} catch (Exception e) {
    producer.abortTransaction();   // 一起回滚
}

> 划重点：Kafka 事务不能跨系统回滚你本地的数据库事务，千万别以为 `producer.abortTransaction()` 能把 MySQL 的 insert 也给回滚了，那是想多了嗷！

五、消息积压——你的队列堵车了，怎么办？

消息积压是线上最常见的问题之一了。咱就是说，哪个搞 Kafka 的没遇到过几次"消费跟不上生产"的紧急情况捏？

怎样定位积压

第一步当然是看看积压到底有多严重：

# 查看消费者组的消费情况
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --group your-group --describe

看 `LAG` 那一列，如果动不动就几十万上百万，那恭喜你，喜提积压大礼包一份 🎁。

积压的几种常见原因和解决套路

1. 消费者太慢了

最直接的原因就是消费逻辑里面有慢操作，比如一条消息处理要 5 秒钟，那一个 consumer 一秒就处理 0.2 条，不积压才见鬼了。

解决：加消费者数量。Kafka 一个 topic 有多少分区，最多就有多少个消费者并行消费。所以如果分区数是 3，那你开 10 个消费者也没用，只有 3 个在干活。

# 先把分区数扩上去（这个操作要注意，已经存在的 key 路由可能会变）
kafka-topics.sh --bootstrap-server localhost:9092 \
    --alter --topic your-topic --partitions 10

然后 consumer 开够 10 个，理论上处理能力就翻了 3 倍多。

// 增加消费者并发线程数
@KafkaListener(topics = "your-topic", concurrency = "10")
public void consume(String msg) {
    // 处理逻辑
}

2. 消费逻辑里有锁竞争或 IO 瓶颈

有些朋友写的消费逻辑里面，每条消息都要去查一次外部 API，而那个 API 每秒只能处理 100 个请求——那你开 10 个 consumer 也没用，瓶颈在那边。

这时候要做的就是优化消费逻辑本身：

- 批量查询代替逐条查询（攒够一批消息一起查）

- 能加缓存的地方加个缓存

- 没必要的同步操作改成异步

// 不好的写法：每条消息都查一次数据库
@KafkaListener(topics = "order-topic")
public void consume(String msg) {
    User user = userMapper.selectById(msg.getUserId());  // 每次都查！
    // ...
}

// 改进：攒一批批量查
@KafkaListener(topics = "order-topic")
public void consume(List<String> messages) {
    List<Long> userIds = messages.stream()
        .map(m -> JSON.parseObject(m, OrderMsg.class).getUserId())
        .collect(Collectors.toList());
    // 一次批量查询
    List<User> users = userMapper.selectBatchIds(userIds);
    // 然后在内存里做匹配处理
    // ...
}

3. 数据倾斜，某个分区特别忙

如果你发现大多数消费者都很闲，就一两个跑满了，那大概率是数据倾斜了——某个分区的消息特别多。

这时候得检查一下生产者的分区策略，看是不是 key 设得有问题：

// 如果以 userId 作为 key，某个大客户的请求量特别大，那消息就全堆一个分区了
producer.send(new ProducerRecord<>("topic", userId.toString(), message));

// 如果消息之间没有严格的顺序要求，可以不设 key，让它轮询均匀分布
producer.send(new ProducerRecord<>("topic", null, message));

4. 终极方案：紧急情况下跳过非关键处理

如果线上已经炸了，先止损再说。临时写个快速消费者把积压的消息吞掉，只做最关键的处理：

@KafkaListener(topics = "your-topic")
public void emergencyConsume(String msg) {
    // 只做必须的操作，能跳过的全跳过
    OrderMsg orderMsg = JSON.parseObject(msg, OrderMsg.class);
    // 只记录关键字段到数据库，别的不干了
    orderEmergencyMapper.insertMinimal(orderMsg);
    // 日志、通知、统计等先全砍掉
}

等积压消掉了再切回正常的消费者。

积压预防比救火更重要

平时就得搭好监控：

// 日常监控 LAG 指标，设置告警阈值
// 比如 LAG > 10000 并且持续 5 分钟就告警
// 这个可以在 Prometheus + Grafana 里搞定

平时多盯着，别等用户投诉了才发现队列炸了——那时候黄花菜都凉了。

总结

Kafka 作为一个老牌消息队列，在这些经典场景下还是相当能打的：

- 异步解耦：让服务之间各玩各的，新增下游不改上游

- 削峰填谷：用缓冲应对瞬时高并发，保护后端脆弱的数据库

- 延迟消息：有点勉强，但能绕弯子实现（真刚需建议上 RocketMQ）

- 事务消息：配合本地消息表做最终一致性，稳如老狗

- 积压处理：加分区加消费力 + 批量优化 + 监控告警，三件套走起

当然，上面很多方案是走"最终一致性"路线的，如果你的业务要求强一致性（比如银行转账），那就得多掂量掂量了，不能啥都往 Kafka 里怼。

以上是个人的一些经验分享，如果哪里有错误的地方也请大佬们指出，咱一起进步！

                                                                                                                     本文完结撒花！！！🎉

---

觉得有用的话，点个赞呗～