rust - 在 Rust 消费者中消费多个 Kafka 主题

Question

我正在尝试使用 Rust 消费者读取多个主题。这是我现在的代码:

extern crate kafka;
use kafka::client::KafkaClient;
use kafka::consumer::Consumer;
use kafka::utils;
fn main(){
    let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
    let res = client.load_metadata_all();
    let topics = client.topic_partitions.keys().cloned().collect(); 
    let offsets = client.fetch_offsets(topics, -1);
    for topic in &topics {
    let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "topic".to_owned()).partition(0);
    let mut messagedata = 0;
    for msg in con {
        println!("{}", str::from_utf8(&msg.message).unwrap().to_string());
    }
  }
}

错误如下:

    src/main.rs:201:19: 201:25 error: use of moved value: `topics` [E0382]
src/main.rs:201     for topic in &topics {
                                  ^~~~~~
    note: in expansion of for loop expansion
    src/main.rs:201:5: 210:6 note: expansion site
    src/main.rs:167:40: 167:46 note: `topics` moved here because it has type `collections::vec::Vec<collections::string::String>`, which is non-copyable
    src/main.rs:167     let offsets = client.fetch_offsets(topics, -1);
                                                       ^~~~~~
    src/main.rs:203:37: 203:43 error: use of moved value: `client` [E0382]
    src/main.rs:203     let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "topicname".to_owned()).partition(0);
                                                    ^~~~~~
    note: in expansion of for loop expansion
    src/main.rs:201:5: 210:6 note: expansion site
    note: `client` was previously moved here because it has type     `kafka::client::KafkaClient`, which is non-copyable
    error: aborting due to 2 previous errors

为了更好地解释我的问题，这里是我仅针对一个主题的部分可行代码:

let mut con = Consumer::new(client, "test-consumer-group".to_owned(), "testtopic".to_owned()).partition(0);

for msg in con {
    println!("{}", str::from_utf8(&msg.message).unwrap().to_string());
}

我测试了 fetch_message 函数，它适用于多个主题，但我得到的结果 (msgs) 是 Topicmessage，我不知道如何从 Topicmessage 获取消息。

let msgs = client.fetch_messages_multi(vec!(utils::TopicPartitionOffset{
                                            topic: "topic1".to_string(),
                                            partition: 0,
                                            offset: 0 //from the begining
                                            },
                                        utils::TopicPartitionOffset{
                                            topic: "topic2".to_string(),
                                            partition: 0,
                                            offset: 0
                                        },
                                        utils::TopicPartitionOffset{
                                            topic: "topic3".to_string(),
                                            partition: 0,
                                            offset: 0
                                        }));
for msg in msgs{
    println!("{}", msg);
}

Answer 1

最后，我像这样更改了代码并且它起作用了。

let msgs = client.fetch_messages_multi(...).unwrap();
for msg in msgs{
     println!("{}", msg.message);
}