Rust结合RabbitMQ异步函数调用异常问题排查

我是Rust新手，在使用异步函数向RabbitMQ队列发送消息时遇到问题：拆分函数实现时消息永远无法发送，但合并成单个函数就一切正常，想知道错误原因。

拆分函数的实现（无法发送消息） #

获取Channel的函数 #

// 获取Channel
async fn get_amqp_channel() -> Channel {
    let connection_arguments = OpenConnectionArguments::new(RABBIT_SERVER_URL, PORT, USER, PASSWORD);
    let connection = Connection::open(&connection_arguments).await.unwrap();
    return connection.open_channel(None).await.unwrap();
}

发送消息的函数 #

// 发送消息
async fn send_amqp_message(channel: &Channel, routing_key: &str, message: String) {
    let publish_arguments = BasicPublishArguments::new(EXCHANGE, routing_key);
    channel.basic_publish(BasicProperties::default(), message.into_bytes(), publish_arguments).await.unwrap();
}

调用代码 #

fn send_command() {
    // 构建消息
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(send_message(message_type, serde_json::to_string(&message).unwrap()));
}

async fn send_message(message_type : String, message : String) {
    let channel = get_amqp_channel().await;
    send_amqp_message(&channel, get_routing_key(message_type).as_str(), message).await;
}

合并后的实现（正常工作） #

async fn send_message(message_type : String, message : String) {
    // 获取Channel
    let connection_arguments = OpenConnectionArguments::new(RABBIT_SERVER_URL, PORT, USER, PASSWORD);
    let connection = Connection::open(&connection_arguments).await.unwrap();
    let channel = connection.open_channel(None).await.unwrap();
    
    // 发送消息
    let publish_arguments = BasicPublishArguments::new(EXCHANGE, get_routing_key(message_type).as_str());
    channel.basic_publish(BasicProperties::default(), message.into_bytes(), publish_arguments).await.unwrap();
}

问题原因 #

核心问题是**Connection的生命周期过短**：在拆分的get_amqp_channel函数中，Connection是局部变量，当函数返回Channel后，Connection会被立即销毁。而RabbitMQ的Channel完全依赖于它的父Connection，一旦Connection被drop，Channel就会失效，后续的消息发送操作自然无法执行。

合并后的代码中，Connection和Channel处于同一个作用域，Connection会一直存活到send_message函数执行完毕，因此Channel能正常完成消息发送。

修正方法 #

调整get_amqp_channel函数，让它同时返回Connection和Channel，确保Connection的生命周期覆盖Channel的使用周期：

// 修改后：同时返回Connection和Channel
async fn get_amqp_connection_and_channel() -> (Connection, Channel) {
    let connection_arguments = OpenConnectionArguments::new(RABBIT_SERVER_URL, PORT, USER, PASSWORD);
    let connection = Connection::open(&connection_arguments).await.unwrap();
    let channel = connection.open_channel(None).await.unwrap();
    (connection, channel)
}

// 调整调用逻辑
async fn send_message(message_type : String, message : String) {
    let (connection, channel) = get_amqp_connection_and_channel().await;
    send_amqp_message(&channel, get_routing_key(message_type).as_str(), message).await;
    // 此时connection会在函数结束后被drop，消息已发送完成
}

这样Connection会和channel一起在send_message的作用域内存活，直到消息发送完成后才被销毁，Channel就能正常完成消息发送操作。

内容的提问来源于stack exchange，提问作者JuanN