orca-mq — 统一消息队列抽象层

> 作者：小黎的部落 | 日期：2026-10-30 | 系列：Rust 基础设施工具集

一、消息队列的碎片化困境

在企业级开发中，消息队列是基础设施的核心组件。但不同团队、不同项目可能使用不同的 MQ：

• 初创团队用 RabbitMQ，简单易上手
• 大数据团队用 Kafka，高吞吐低延迟
• 云原生团队用 Pulsar，存算分离

问题来了：每换一个 MQ，业务代码就要重写一遍。

orca-mq 通过统一的 Producer/Consumer trait 解决了这个问题。

二、核心设计

2.1 统一 Trait

// 消息 trait：所有消息类型必须实现
pub trait Message {
    fn topic(&self) -> &str;
    fn key(&self) -> Option<&str> { None }
}

// 生产者 trait
#[async_trait]
pub trait MqProducer {
    async fn send(&self, msg: &dyn Message) -> MqResult<()>;
    async fn send_batch(&self, msgs: &[&dyn Message]) -> MqResult<()>;
}

// 消费者 trait
#[async_trait]
pub trait MqConsumer {
    async fn subscribe(&self, topic: &str, handler: Box<dyn MessageHandler>) -> MqResult<()>;
    async fn start(&self) -> MqResult<()>;
}

2.2 类型安全的消息定义

use orca_mq::Message;
use serde::{Serialize, Deserialize};

#[derive(Serialize, Deserialize)]
struct OrderMessage {
    order_id: String,
    amount: f64,
    items: Vec<String>,
}

impl Message for OrderMessage {
    fn topic(&self) -> &str { "orders" }
    fn key(&self) -> Option<&str> { Some(&self.order_id) }
}

2.3 生产者示例

use orca_mq::{MessageProducer, ProducerConfig};

let config = ProducerConfig::new("localhost:9092")
    .with_ack_timeout(Duration::from_secs(5))
    .with_compression(true);

let producer = MessageProducer::new(config).await?;

let msg = OrderMessage {
    order_id: "ORD-001".into(),
    amount: 99.99,
    items: vec!["item-1".into()],
};

producer.send(&msg).await?;

2.4 消费者示例

use orca_mq::{MessageConsumer, ConsumerConfig, MessageHandler};

struct OrderHandler;

#[async_trait]
impl MessageHandler<OrderMessage> for OrderHandler {
    async fn handle(&self, msg: &OrderMessage) -> Result<(), String> {
        println!("处理订单: {}, 金额: {}", msg.order_id, msg.amount);
        Ok(())
    }
}

let config = ConsumerConfig::new("localhost:9092", "order-group")
    .with_auto_commit(false);

let consumer = MessageConsumer::new(config).await?;
consumer.subscribe("orders", OrderHandler).await?;
consumer.start().await?;

三、后端切换

从 Kafka 切换到 RabbitMQ，只需修改配置：

// Kafka
let config = ProducerConfig::new("kafka://localhost:9092");

// RabbitMQ
let config = ProducerConfig::new("amqp://localhost:5672");

// 业务代码完全不变
let producer = MessageProducer::new(config).await?;
producer.send(&msg).await?;

四、与其他方案对比

五、总结

orca-mq 的设计哲学是：业务代码不应该关心底层是 Kafka 还是 RabbitMQ。通过统一的 trait 抽象和类型安全的消息定义，它让消息队列的切换变得像换一个配置参数一样简单。

六、下载

下载 orca-mq.tar.gz (3.9 KB)