第75章消息队列

课程目录第75章消息队列

消息队列：RabbitMQ/Kafka集成、生产者消费者、协程与消息队列

消息队列这东西，说白了就是系统之间的「缓冲区」。我刚开始接触微服务时，总觉得服务之间直接调API就完事了，干嘛还要多此一举加个消息中间件？直到有一次线上服务雪崩，A服务挂了导致B服务也挂了，连锁反应搞得整个系统瘫痪……嗯，从那以后，我再也不敢轻视消息队列了。

今天咱们聊聊Kotlin项目中怎么集成RabbitMQ和Kafka，以及如何用协程优雅地处理消息的生产与消费。

为什么需要消息队列？

先想一个问题：你的用户注册后，需要发邮件、发短信、更新推荐系统、同步到第三方平台……这些操作如果同步执行，用户得等多久？

消息队列的核心价值就三点：

解耦：生产者只管发消息，不用关心谁在消费
削峰填谷：突发流量先存到队列里，消费者慢慢处理
异步提速：用户请求立刻返回，后台慢慢干活

我的经验：在电商项目中，秒杀场景下消息队列是标配。我曾经见过一个团队没加消息队列，结果大促时数据库连接池直接被冲垮，恢复用了两个小时……

RabbitMQ vs Kafka：怎么选？

很多新手会纠结这个问题。我个人的习惯是：

对比维度	RabbitMQ	Kafka
消息模型	交换机+队列	Topic+分区
吞吐量	万级/秒	百万级/秒
消息可靠性	高（支持确认机制）	高（支持副本机制）
适用场景	任务调度、业务解耦	日志收集、流处理、大数据
学习成本	较低	中等

说白了，如果你的业务需要灵活的路由、延迟队列、死信队列这些功能，RabbitMQ更合适。如果你要处理海量日志或实时数据流，Kafka是更好的选择。

Kotlin集成RabbitMQ

咱们先看RabbitMQ。在Kotlin项目中，我推荐用官方的Java客户端，配合协程来封装。

生产者示例

// 使用协程发送消息
suspend fun sendMessage(exchange: String, routingKey: String, message: String) {
    withContext(Dispatchers.IO) {
        val factory = ConnectionFactory().apply {
            host = "localhost"
            port = 5672
            username = "guest"
            password = "guest"
        }
        factory.newConnection().use { connection ->
            connection.createChannel().use { channel ->
                channel.exchangeDeclare(exchange, "direct", true)
                channel.basicPublish(exchange, routingKey, null, message.toByteArray())
                println("消息已发送: $message")
            }
        }
    }
}

小技巧：用use块可以自动关闭资源，避免内存泄漏。我在早期项目里吃过这个亏，连接没关导致RabbitMQ服务端报连接数超限。

消费者示例

// 协程消费者
fun startConsumer(queueName: String) {
    CoroutineScope(Dispatchers.IO).launch {
        val factory = ConnectionFactory().apply {
            host = "localhost"
        }
        factory.newConnection().use { connection ->
            connection.createChannel().use { channel ->
                channel.queueDeclare(queueName, true, false, false, null)
                val consumer = object : DefaultConsumer(channel) {
                    override fun handleDelivery(
                        consumerTag: String,
                        envelope: Envelope,
                        properties: BasicProperties,
                        body: ByteArray
                    ) {
                        val message = String(body, Charsets.UTF_8)
                        println("收到消息: $message")
                        // 手动确认
                        channel.basicAck(envelope.deliveryTag, false)
                    }
                }
                channel.basicConsume(queueName, false, consumer)
                // 保持协程运行
                delay(Long.MAX_VALUE)
            }
        }
    }
}

注意：basicConsume的第二个参数设为false，表示手动确认。我曾经遇到过自动确认导致消息丢失的情况——消费者还没处理完，消息就被标记为已消费了。

Kotlin集成Kafka

Kafka的集成相对简单一些，官方提供了Kafka Clients库。我个人习惯用Spring Kafka，但如果你不想引入Spring，直接用原生客户端也行。

生产者示例

// Kafka生产者
class KafkaProducer(private val bootstrapServers: String) {
    private val producer = KafkaProducer<String, String>(
        mapOf(
            ProducerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
            ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
            ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG to StringSerializer::class.java,
            ProducerConfig.ACKS_CONFIG to "all"  // 等待所有副本确认
        )
    )

    suspend fun send(topic: String, key: String, value: String) {
        withContext(Dispatchers.IO) {
            val future = producer.send(ProducerRecord(topic, key, value))
            // 等待发送结果
            val metadata = future.get()
            println("消息发送到分区: ${metadata.partition()}, 偏移量: ${metadata.offset()}")
        }
    }

    fun close() {
        producer.close()
    }
}

消费者示例

// Kafka消费者 + 协程
class KafkaConsumer(private val bootstrapServers: String, private val groupId: String) {
    private val consumer = KafkaConsumer<String, String>(
        mapOf(
            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG to bootstrapServers,
            ConsumerConfig.GROUP_ID_CONFIG to groupId,
            ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
            ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG to StringDeserializer::class.java,
            ConsumerConfig.AUTO_OFFSET_RESET_CONFIG to "earliest"
        )
    )

    fun subscribe(topics: List<String>) {
        consumer.subscribe(topics)
    }

    suspend fun poll(duration: Long = 1000) {
        withContext(Dispatchers.IO) {
            while (true) {
                val records = consumer.poll(Duration.ofMillis(duration))
                for (record in records) {
                    println("收到消息: key=${record.key()}, value=${record.value()}, offset=${record.offset()}")
                    // 处理业务逻辑
                    processMessage(record.value())
                }
                // 手动提交偏移量
                consumer.commitSync()
            }
        }
    }

    private fun processMessage(value: String) {
        // 业务处理
    }

    fun close() {
        consumer.close()
    }
}

避坑指南：我曾经在Kafka消费者中忘记提交偏移量，导致重启后重复消费了大量消息。建议用commitSync或commitAsync手动控制提交时机。

协程与消息队列的完美结合

为什么要把协程和消息队列放在一起讲？因为消息队列的I/O操作天然适合协程。

你想想看：生产者发送消息时要等待网络响应，消费者拉取消息时要阻塞等待。如果用线程池，每个消费者都要占用一个线程，100个消费者就是100个线程，资源浪费严重。

用协程就不一样了：

发送消息时挂起协程，不阻塞线程
消费消息时用Dispatchers.IO调度，线程复用
可以轻松管理大量消费者实例

// 批量启动消费者协程
fun startMultipleConsumers(count: Int, topic: String) {
    repeat(count) { index ->
        CoroutineScope(Dispatchers.IO + SupervisorJob()).launch {
            val consumer = KafkaConsumer("localhost:9092", "group-$index")
            consumer.subscribe(listOf(topic))
            try {
                consumer.poll()
            } catch (e: Exception) {
                println("消费者 $index 异常: ${e.message}")
            } finally {
                consumer.close()
            }
        }
    }
}

我的习惯：用SupervisorJob让每个消费者独立运行，一个挂了不影响其他消费者。这在生产环境中特别重要。

知识体系总览

下面这张图帮你理清消息队列在Kotlin项目中的核心逻辑：

生产环境注意事项

最后分享几个我在生产环境中踩过的坑：

连接池管理：不要每次发送消息都创建新连接。用连接池复用，RabbitMQ的ConnectionFactory默认会做连接池，但Kafka需要自己管理。
消息幂等性：消费者要做好幂等处理。我遇到过网络重试导致消息重复消费的情况，后来在业务层加了去重逻辑。
监控告警：消息堆积是常见问题。建议监控队列深度，超过阈值就告警。我曾经凌晨三点被电话叫醒，就是因为消息队列堆积了10万条消息没人处理。
优雅关闭：应用关闭时要等待正在处理的消息完成。用Runtime.getRuntime().addShutdownHook注册关闭回调。

重要提醒：协程虽然轻量，但也不是无限的。每个消费者启动一个协程，如果消费者数量达到几千个，建议用Channel或Flow做协程间的消息分发，而不是每个消费者一个独立协程。

好了，消息队列的集成就聊到这儿。记住一个原则：能用消息队列解耦的地方，就别硬写同步调用。你的系统会感谢你的。

📚 课程目录

消息队列：RabbitMQ/Kafka集成、生产者消费者、协程与消息队列

为什么需要消息队列？

RabbitMQ vs Kafka：怎么选？

Kotlin集成RabbitMQ

生产者示例

消费者示例

Kotlin集成Kafka

生产者示例

消费者示例

协程与消息队列的完美结合

知识体系总览

生产环境注意事项