Rust对接Kafka时如何处理长时间运行的操作？

我正在开发一个Rust应用，从Kafka读取数据并归档到磁盘。部分场景下，数据格式转换和磁盘写入的函数运行时间很长（最长可达数分钟），这会导致Kafka消费者崩溃——下次轮询时会发现已被Broker断开连接（默认超时30秒）。

之前用Python解决过这个问题：启动独立线程，暂停消费者后持续轮询，等待事件触发后恢复消费者并退出；主线程负责启动该线程、触发事件并等待线程结束。

但在Rust中实现时代码变得混乱，需要用Arc包裹Mutex的对象，还要处理作用域释放Mutex，维护难度很高。主要是对Rust多线程机制理解不够——这类操作在C++里很容易实现，没有Rust借用检查器的约束。

单线程崩溃示例代码 #

以下是单线程实现，调用process_message时会崩溃（假设未手动延长Kafka消费者默认轮询超时，函数退出后消费者会失败）：

use core::str;

use rdkafka::consumer::Consumer;
use rdkafka::Message;

fn process_message(message: &str) {
    println!("message: {message}");
    println!("sleep 60");
    std::thread::sleep(std::time::Duration::from_secs(60));
    println!("wake");
}

fn main() -> Result<(), ()> {

    let config =
        rdkafka::config::ClientConfig::new()
        .set("bootstrap.servers", "192.168.0.239:9092")
        .set("client.id", "html_source_div_archiver.rust")
        .set("group.id", "html_source_div_archiver.rust")
        .set("enable.auto.commit", "false")
        .set("auto.offset.reset", "earliest")
        .set("isolation.level", "read_committed")
        .to_owned();

    let consumer: rdkafka::consumer::BaseConsumer =
        config
        .create()
        .expect("invalid client configuration");

    let topic_html_source_div = "example_topic";
    let topics = [topic_html_source_div];
    consumer.subscribe(&topics).expect("failed to subscribe");

    loop {

        let poll_timeout = std::time::Duration::from_secs(10);
        let optional_message = consumer.poll(poll_timeout);

        match optional_message {
            Some(maybe_message) => {

                let message = maybe_message.expect("some message failure");

                let optional_payload = message.payload();
                match optional_payload {
                    Some(payload) => {

                        let message_str = str::from_utf8(payload).expect("invalid UTF-8");
                        process_message(message_str);

                        consumer
                            .commit_consumer_state(rdkafka::consumer::CommitMode::Sync)
                            .expect("failed to commit consumer state");
                    }

                    None => {
                        panic!("payload is None");
                    }
                }
            }

            None => {
                println!("waiting for messages");
            }
        }
    }
}

基于Python思路的Rust尝试 #

之前Python的解决方案核心是consumer_poll_loop函数，完成以下操作：

• 暂停消费者
• 持续循环，在等待事件触发的同时轮询已暂停的消费者
• 恢复消费者

以下是我尝试用Rust实现的对应函数：

use std::sync::atomic::AtomicBool;
use std::sync::atomic;
use std::sync::Arc;
use std::sync::Mutex;

fn consumer_poll_loop(
    event_thread_terminate: Arc<AtomicBool>,
    consumer: Arc<Mutex<rdkafka::consumer::BaseConsumer>>,
) {
    let consumer = consumer.lock().expect("failed to lock consumer mutex");

    let topic_partition_list = consumer.assignment().expect("consumer assignment failed");
    consumer.pause(&topic_partition_list).expect("consumer pause failed");

    loop {
        let poll_timeout = std::time::Duration::from_secs(10);
        let message = consumer.poll(poll_timeout);

        if message.is_some() {
            panic!("message is not None");
        }

        if event_thread_terminate.load(atomic::Ordering::Relaxed) {
            break;
        }
    }

    consumer.resume(&topic_partition_list).expect("consumer resume failed");
}

注：此函数中无需手动释放Mutex，因为该线程运行时其他线程不应访问消费者。

我已能启动该函数，但必须使用闭包，不确定是否有必要；后续在main函数中实现逻辑时代码变得一团糟，显然实现方式有误。

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