reqwest 作为客户端和 axum 作为服务器的问题（axum 在 1k req/s 时爆炸，端口使用错误）

Question

我有问题，为什么发送时出现此错误request

\n

Error: reqwest::Error {\n    kind: Request, \n    url: Url { \n        scheme: "http", \n        cannot_be_a_base: false, \n        username: "", \n        password: None, \n        host: Some(Ipv4(127.0.0.1)), \n        port: Some(3000), \n        path: "/message", \n        query: None, \n        fragment: None \n    }, \n    source: hyper::Error(\n        Connect, \n        ConnectError(\n            "tcp connect error", \n            Os { \n                code: 10048, \n                kind: AddrInUse, \n                message: "Only one usage of each socket address (protocol/network address/port) is normally permitted." \n            }\n        )\n    )\n}\n

\n

它以“随机”速率发生，例如 15-45 秒后cargo run。我用作axum服务器和reqwest请求客户端。这是来源。第一段代码不是多线程的，它只是在循环中运行 ( for message in channel)，没有 ( async) 生成，第二段代码正在解析消息并将其一些上下文写入数据库，我正在使用 SQLite 作为sqlx数据库。错误消息也非常令人困惑，除了在端口 3000 上写入数据库的应用程序之外，我不提供任何服务，所以这怎么可能。我也在使用reqwest客户端，pool_max_idle_per_host设置为 0，因为这是我在这个 GitHub 问题hyper::Error(IncompleteMessage): connection closed before message completed.中描述的解决方法。

\n

//sending at rate of ~1000 per second\npub async fn send_message_to_extractor(\n    message: DiscordMessage,\n    client: &reqwest::Client,\n) -> Result<(), reqwest::Error> {\n    //info!("sending message to extractor: {:?}", message);\n    let mut message = message;\n    message.guild_id = Some("1234".to_owned());\n    let request = client\n        .post(EXTRACTOR_URL)\n        .json(&message)\n        .send()\n        .await;\n    match request {\n        Ok(_) => {\n            // returned\n            Ok(())\n        },\n        Err(e) => Err(e),\n    }\n}\n

\n

pub async fn handle_discord_message(pool: Extension<SqlitePool>, Json(payload): Json<DiscordMessage>) -> impl IntoResponse {\n    ...\n    // Parsing message about ~400ms of work\n    ...\n    return StatusCode::OK;\n}\n

\n

    // Code that calls send_message_to_extractor functions which fails after 15-45 sec\n    let reqwest_client = reqwest::Client::builder()\n        .pool_max_idle_per_host(0)\n        .build()?;\n    while let Some(row) = stream.try_next().await? {\n        let data: Vec<u8> = row.try_get("data")?;\n        // turn data to string\n        let data_string = if COMPRESSION {\n            decode_reader(data)?\n        } else {\n            String::from_utf8(data)?\n        };\n        // turn string to discord message\n        let discord_message: DiscordMessage = serde_json::from_str(&data_string)?;\n        // send message to extractor\n        //println!("sending message to extractor: {:?}", discord_message);\n        send_message_to_extractor(\n            discord_message,\n            &reqwest_client\n        ).await?;\n        bar.inc(1);\n        // Do something with the data vector...\n    }\n\n    Ok(())\n

\n

// main.rs\n    let pool = SqlitePoolOptions::new()\n        .max_connections(50)\n        .connect(&DATABASE_URL)\n        .await?;\n\n    let app = Router::new()\n        .route("/message", post(routes::handle_discord_message))\n        .layer(Extension(pool));\n\n    let addr = SocketAddr::from(([127, 0, 0, 1], 3000));\n    tracing::info!("listening on {}", addr);\n    axum::Server::bind(&addr)\n        .serve(app.into_make_service())\n        .await\n        .unwrap();\n\n    Ok(())\n

\n

我尝试过弄乱保持活动标头和不同的客户端设置。唯一改变的是程序崩溃的时间，大约为 \xc2\xb130 秒。除了端口 3000 上的应用程序之外，我没有运行任何程序axum。

\n