第19章消息队列集成

课程目录第19章消息队列集成

消息队列集成：RabbitMQ/Kafka集成、生产者消费者模式、异步任务处理

说到后端架构，消息队列这东西，我敢说但凡你做过稍微有点规模的系统，迟早得碰上。我记得刚入行那会儿，总觉得消息队列是“大厂才用的东西”，自己写个同步接口不就完事了？后来线上出了几次事故，我才明白——异步才是解耦的王道。

这一章，咱们就聊聊在Ktor里怎么集成消息队列。我会重点讲RabbitMQ和Kafka，这两个是目前最主流的方案。说白了，一个偏传统企业级，一个偏大数据流式处理。你根据场景选就行。

  核心概念一句话：消息队列就是让生产者把消息扔进队列，消费者异步去拿。生产者和消费者彼此不直接依赖，系统因此变得灵活、可扩展。

为什么要在Ktor里用消息队列？

你想想看，一个Android后端的API，如果每次请求都要同步处理所有逻辑——比如发短信、写日志、推送通知——那响应时间肯定爆炸。更糟糕的是，如果某个下游服务挂了，整个请求就失败了。

用消息队列之后，API只需要把任务塞进队列，立刻返回“已接收”。后台的消费者慢慢处理。这样用户体验好，系统也稳定。我在项目中遇到过好几次，数据库写入压力太大导致接口超时，后来引入消息队列做削峰填谷，问题就解决了。

RabbitMQ集成：经典的生产者消费者

RabbitMQ是个老牌消息中间件，基于AMQP协议。它的核心概念就几个：交换机（Exchange）、队列（Queue）、绑定（Binding）。生产者发消息到交换机，交换机根据路由规则把消息投递到对应的队列，消费者从队列里取消息。

在Ktor里集成RabbitMQ，我一般用官方的Java客户端。先加依赖：

// build.gradle.kts
implementation("com.rabbitmq:amqp-client:5.20.0")

然后写一个连接管理类。我个人习惯用单例模式，避免反复创建连接：

object RabbitMQManager {
    private val factory = ConnectionFactory().apply {
        host = "localhost"
        port = 5672
        username = "guest"
        password = "guest"
    }

    val connection: Connection by lazy { factory.newConnection() }
    val channel: Channel by lazy { connection.createChannel() }
}

生产者这边，其实就是声明一个队列，然后发布消息：

fun sendMessage(queueName: String, message: String) {
    val channel = RabbitMQManager.channel
    channel.queueDeclare(queueName, durable = true, exclusive = false, autoDelete = false, null)
    channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.toByteArray())
    println(" [x] Sent '$message'")
}

消费者呢，需要实现一个回调接口。这里要注意，消费者最好在单独的协程里运行，别阻塞Ktor的事件循环：

fun startConsumer(queueName: String) {
    val channel = RabbitMQManager.channel
    channel.queueDeclare(queueName, durable = true, exclusive = false, autoDelete = false, null)

    val deliverCallback = DeliverCallback { _, delivery ->
        val message = String(delivery.body, Charsets.UTF_8)
        println(" [x] Received '$message'")
        // 在这里处理业务逻辑
        channel.basicAck(delivery.envelope.deliveryTag, false)
    }

    channel.basicConsume(queueName, false, deliverCallback, _ -> {})
}

避坑指南：我曾经在生产环境忘了设置消息持久化，结果RabbitMQ重启后所有消息都丢了。血的教训——MessageProperties.PERSISTENT_TEXT_PLAIN 这个参数一定要加上。

Kafka集成：高吞吐的流式处理

Kafka和RabbitMQ的思路不太一样。Kafka不是队列，而是日志（Log）。消息按主题（Topic）分类，每个主题有多个分区（Partition）。消费者读取消息时，可以指定偏移量（Offset），所以Kafka天然支持消息重播。

如果你的场景是日志收集、事件溯源、或者需要处理海量数据流，Kafka是更好的选择。我在做一个物联网项目时，设备每秒上报上万条数据，用RabbitMQ直接扛不住，换成Kafka后稳如老狗。

Ktor集成Kafka，我推荐用官方的Kafka客户端：

// build.gradle.kts
implementation("org.apache.kafka:kafka-clients:3.6.0")

生产者配置：

val producerProps = Properties().apply {
    put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
    put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
    put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
    put(ProducerConfig.ACKS_CONFIG, "all") // 等待所有副本确认
}

val producer = KafkaProducer<String, String>(producerProps)

fun sendToKafka(topic: String, key: String, value: String) {
    producer.send(ProducerRecord(topic, key, value)) { metadata, exception ->
        if (exception != null) {
            System.err.println("发送失败: ${exception.message}")
        } else {
            println("发送成功: partition=${metadata.partition()}, offset=${metadata.offset()}")
        }
    }
}

消费者这边，需要订阅主题并轮询消息。Kafka的消费者是拉模式（Pull），不像RabbitMQ是推模式（Push）：

fun startKafkaConsumer(topic: String) {
    val consumerProps = Properties().apply {
        put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092")
        put(ConsumerConfig.GROUP_ID_CONFIG, "ktor-consumer-group")
        put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
        put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
        put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
    }

    val consumer = KafkaConsumer<String, String>(consumerProps)
    consumer.subscribe(listOf(topic))

    // 在协程里循环拉取
    GlobalScope.launch {
        while (true) {
            val records = consumer.poll(Duration.ofMillis(100))
            for (record in records) {
                println("收到消息: key=${record.key()}, value=${record.value()}, offset=${record.offset()}")
                // 处理业务逻辑
            }
        }
    }
}

注意：Kafka消费者必须设置 GROUP_ID_CONFIG，否则会报错。同一个组内的消费者会负载均衡地消费分区。如果你希望每个消费者都收到全量消息，那就用不同的组ID。

异步任务处理：让Ktor不阻塞

消息队列的核心价值，就是实现异步任务处理。在Ktor里，我通常这样设计：

API层：接收请求，把任务封装成消息，扔进队列，立即返回202 Accepted。
消费者层：在后台协程里消费消息，执行耗时操作（如发邮件、生成报表、调用第三方API）。
结果通知：如果客户端需要知道结果，可以用WebSocket或者轮询另一个状态查询接口。

举个例子，用户注册后需要发送欢迎邮件。同步处理的话，接口可能要等好几秒。用消息队列就变成了：

fun Application.registerModule() {
    routing {
        post("/register") {
            val user = call.receive<User>()
            // 保存用户到数据库...
            // 发送消息到邮件队列
            sendMessage("email.queue", "welcome:${user.email}")
            call.respond(HttpStatusCode.Accepted, "注册成功，邮件稍后发送")
        }
    }
}

消费者那边收到消息后，再真正调用邮件服务。这样API的响应时间就从秒级降到了毫秒级。

RabbitMQ vs Kafka：怎么选？

很多新手会纠结这个问题。我直接给个对照表，你根据需求选：

维度	RabbitMQ	Kafka
消息模型	队列 + 交换机	主题 + 分区 + 偏移量
吞吐量	万级/秒	百万级/秒
消息顺序	单队列有序	单分区有序
消息持久化	支持，但重启恢复较慢	天生持久化，基于磁盘日志
典型场景	任务调度、RPC、微服务解耦	日志收集、事件溯源、流处理
运维复杂度	较低，自带管理界面	较高，需要ZooKeeper/KRaft

我个人建议：如果你的消息量不大（日均百万以下），且需要灵活的路由规则，选RabbitMQ。如果你要处理海量数据流，或者需要消息重播，选Kafka。

消息队列的常见坑

嗯，这里我总结几个自己踩过的坑，你注意避开：

消息丢失：生产者没开启确认机制，消费者没做手动ACK。解决方案是开启 publisher-confirms 和 manual ack。
重复消费：网络抖动导致消费者ACK没送达，消息被重新投递。解决方案是让消费者实现幂等性——比如用消息ID去重。
消息堆积：消费者处理速度跟不上生产者。解决方案是增加消费者实例，或者优化处理逻辑。
连接泄漏：每次请求都创建新的连接，导致RabbitMQ连接数爆满。解决方案是复用连接，用连接池。

我的经验：在Ktor里启动消费者时，最好用 application.environment.monitor.subscribe(ApplicationStarted) 事件，确保应用完全启动后再开始消费。否则可能出现数据库连接还没初始化，消费者就开始处理消息的情况。

知识体系总览

下面这张图，帮你理清消息队列在Ktor后端中的位置和作用：

这张图展示了完整的异步处理流程。你看，客户端请求进来，API层立刻返回，消息队列做中转，消费者在后台慢慢处理。整个系统就像一条流水线，每个环节各司其职。

好了，消息队列这块就聊到这儿。记住一句话：能用异步就别同步，能解耦就别硬耦合。你的Ktor后端会感谢你的。

📚 课程目录

消息队列集成：RabbitMQ/Kafka集成、生产者消费者模式、异步任务处理

为什么要在Ktor里用消息队列？

RabbitMQ集成：经典的生产者消费者

Kafka集成：高吞吐的流式处理

异步任务处理：让Ktor不阻塞

RabbitMQ vs Kafka：怎么选？

消息队列的常见坑

知识体系总览