1
我試圖使用Rust消費者從多個主題中讀取。這是代碼我現在有:在Rust消費者中消費多個Kafka主題
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);
}
@Shepmaster謝謝你的擡頭,我修改了我的問題。 – coco
這些問題實際上是Rust的所有權的基礎。請閱讀其他有'使用移動的值'錯誤的stackoverflow問題,並閱讀本書的所有權章節:https://doc.rust-lang.org/stable/book/ownership.html –
@ker謝謝你的幫幫我! – coco