Compare commits
1
Commits
v1.0.0
...
tauri-legacy
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
33d0ca10e7 |
@@ -0,0 +1,66 @@
|
|||||||
|
//! 消息流量压测工具:连接 broker 后按指定速率向主题发布消息。
|
||||||
|
//!
|
||||||
|
//! 用法:cargo run --example flood -- <host> <port> <topic> <count> <per_second> [payload_bytes]
|
||||||
|
//! 例:向私有 broker 发 500 条 1KB 消息、每秒 50 条:
|
||||||
|
//! cargo run --example flood -- mqtt.nex-i.cn 1883 mqttx-flood 500 50 1024
|
||||||
|
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
let args: Vec<String> = std::env::args().collect();
|
||||||
|
if args.len() < 6 {
|
||||||
|
eprintln!("usage: flood <host> <port> <topic> <count> <per_second> [payload_bytes]");
|
||||||
|
std::process::exit(1);
|
||||||
|
}
|
||||||
|
let host = args[1].clone();
|
||||||
|
let port: u16 = args[2].parse().unwrap();
|
||||||
|
let topic = args[3].clone();
|
||||||
|
let count: u64 = args[4].parse().unwrap();
|
||||||
|
let per_second: u64 = args[5].parse().unwrap();
|
||||||
|
let payload_size: usize = args.get(6).map(|s| s.parse().unwrap()).unwrap_or(512);
|
||||||
|
|
||||||
|
let rt = tokio::runtime::Builder::new_multi_thread()
|
||||||
|
.worker_threads(2)
|
||||||
|
.enable_all()
|
||||||
|
.build()
|
||||||
|
.unwrap();
|
||||||
|
rt.block_on(async move {
|
||||||
|
let opts = rumqttc::MqttOptions::new(
|
||||||
|
format!("mqttx_flood_{}", &uuid::Uuid::new_v4().simple().to_string()[..8]),
|
||||||
|
rumqttc::Broker::tcp(&host, port),
|
||||||
|
);
|
||||||
|
let (client, mut eventloop) = rumqttc::AsyncClient::builder(opts).build();
|
||||||
|
|
||||||
|
// eventloop 独立持续 poll:请求出队全靠它
|
||||||
|
tokio::spawn(async move {
|
||||||
|
loop {
|
||||||
|
if eventloop.poll().await.is_err() {
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
let payload = vec![b'x'; payload_size];
|
||||||
|
let interval = Duration::from_secs_f64(1.0 / per_second.max(1) as f64);
|
||||||
|
let start = Instant::now();
|
||||||
|
for i in 0..count {
|
||||||
|
let topic_i = format!("{topic}/{i}");
|
||||||
|
let options = rumqttc::PublishOptions::new(rumqttc::QoS::AtMostOnce);
|
||||||
|
if let Err(e) = client.publish(&topic_i, payload.clone(), options).await {
|
||||||
|
eprintln!("publish error at {i}: {e}");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
let next = start + interval * (i as u32 + 1);
|
||||||
|
let now = Instant::now();
|
||||||
|
if next > now {
|
||||||
|
tokio::time::sleep(next - now).await;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
eprintln!(
|
||||||
|
"flooded {count} msgs ({payload_size}B each) in {:.1}s",
|
||||||
|
start.elapsed().as_secs_f64()
|
||||||
|
);
|
||||||
|
// 等网络冲刷
|
||||||
|
tokio::time::sleep(Duration::from_secs(3)).await;
|
||||||
|
});
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user