使用Hyper 1.x通过RPC调用Testnet比特币节点获取Mempool交易遇阻

我需要通过Bitcoin-RPC协议向Testnet上的比特币节点请求内存池（Mempool）交易列表，因需求必须使用底层库Hyper。Hyper 0.14.28版本下代码正常运行，但升级到最新版本后，代码执行卡住，仅打印"A"却无法打印"B"，怀疑服务器未响应。我编程经验有限，希望得到详细解答。

尝试的代码 #

use base64::Engine;
use bytes::Bytes;
use http_body_util::{BodyExt, Full};
use hyper::header::{AUTHORIZATION, CONTENT_TYPE};
use hyper::Request;
use hyper_util::rt::TokioIo;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tokio::net::TcpStream;

#[tokio::main]
async fn main() {
    let response = send_json_rpc_request("getrawmempool", json!([])).await;
    match response {
        Ok(result) => {
            let response_: Result<Vec<String>, ()> =
                match serde_json::from_value(result.result.unwrap()) {
                    Ok(response_inner) => response_inner,
                    Err(_) => {
                        println!("Deserialization error");
                        Err(())
                    }
                };
            println!("The mempool is: {:?}", response_.unwrap());
        }
        Err(error) => {
            println!("Error: {:?}", error);
        }
    }
}

async fn send_json_rpc_request<T: for<'de> Deserialize<'de> + std::fmt::Debug>(
    method: &str,
    params: serde_json::Value,
) -> Result<JsonRpcResult<T>, JsonRpcError> {
    let url = "http://127.0.0.1:18332".parse::<hyper::Uri>().unwrap();
    let host = url.host().expect("uri has no host");
    let port = url.port_u16().unwrap_or(18332);
    let address = format!("{}:{}", host, port);
    let stream = TcpStream::connect(address).await.unwrap();
    let io = TokioIo::new(stream);
    let (mut sender, _conn) = hyper::client::conn::http1::handshake(io).await.unwrap();
    let (username, password) = ("username", "password");
    let request = JsonRpcRequest {
        jsonrpc: "2.0".to_string(),
        method: method.to_string(),
        params,
        id: 1,
    };

    let request_body = serde_json::to_string(&request).unwrap();

    let req = Request::builder()
        .method("POST")
        .uri(url)
        .header(CONTENT_TYPE, "application/json")
        .header(
            AUTHORIZATION,
            format!(
                "Basic {}",
                base64::engine::general_purpose::STANDARD
                    .encode(format!("{}:{}", username, password))
            ),
        )
        .body(Full::<Bytes>::from(request_body))
        .unwrap();

    println!("A");
    let response = sender.send_request(req).await.unwrap();

    println!("B");
    let status = response.status();
    let body = response.into_body().collect().await.unwrap().to_bytes();

    if status.is_success() {
        serde_json::from_slice(&body).unwrap()
    } else {
        match serde_json::from_slice(&body) {
            Ok(error_response) => Err(error_response),
            Err(e) => Err(JsonRpcError {
                code: -1,
                message: format!("Deserialization error {:?}", e),
            }),
        }
    }
}
#[derive(Debug, Serialize)]
struct JsonRpcRequest {
    jsonrpc: String,
    method: String,
    params: serde_json::Value,
    id: u64,
}

#[derive(Debug, Deserialize)]
pub struct JsonRpcResult<T> {
    result: Option<T>,
    error: Option<JsonRpcError>,
    id: u64,
}

#[derive(Debug, Deserialize, Clone)]
pub struct JsonRpcError {
    code: i32,
    message: String,
}

请求体对比 #

最新Hyper版本中的请求体 #

&req = Request {
    method: POST,
    uri: http://127.0.0.1:18332/,
    version: HTTP/1.1,
    headers: {
        "content-type": "application/json",
        "authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
    },
    body: Full {
        data: Some(
            b"{\"jsonrpc\":\"2.0\",\"method\":\"getrawmempool\",\"params\":[],\"id\":1}",
        ),
    },
}

Hyper 0.14.28版本中的请求体（正常运行） #

&req = Request {
    method: POST,
    uri: http://127.0.0.1:18332/,
    version: HTTP/1.1,
    headers: {
        "content-type": "application/json",
        "authorization": "Basic dXNlcm5hbWU6cGFzc3dvcmQ=",
    },
    body: Body(
        Full(
            b"{\"jsonrpc\":\"2.0\",\"method\":\"getrawmempool\",\"params\":[],\"id\":1}",
        ),
    ),
}

比特币节点配置 #

运行的比特币节点为最新master分支，提交哈希为4b1196a9855dcd188a24f393aa2fa21e2d61f061，配置如下：

[test]
testnet=1
server=1
datadir=<path to your testnet blockchain>
rpcuser=username
rpcpassword=password
rpcport=18332

问题原因及解决方案 #

原因 #

Hyper 1.x版本中，http1::handshake返回的conn是一个需要持续运行的Future，负责处理连接上的IO事件（包括接收响应）。旧版本Hyper会自动处理这个逻辑，但新版本需要显式驱动该Future，否则请求发送后无法接收服务器响应，导致代码卡在send_request处。

解决方案 #

在获取sender和conn后，使用tokio::spawn启动一个任务来运行conn的Future。修改send_json_rpc_request函数中的相关代码：

// 原代码
let (mut sender, _conn) = hyper::client::conn::http1::handshake(io).await.unwrap();

// 修改为：
let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await.unwrap();
// 启动任务处理连接IO
tokio::spawn(async move {
    if let Err(e) = conn.await {
        eprintln!("Connection error: {}", e);
    }
});

此外，建议显式设置Host请求头（Hyper通常会自动添加，但显式设置更稳妥）：

let req = Request::builder()
    // ... 其他配置
    .header("Host", host)
    // ... 其他配置
    .body(Full::<Bytes>::from(request_body))
    .unwrap();

完整修改后的核心函数片段 #

async fn send_json_rpc_request<T: for<'de> Deserialize<'de> + std::fmt::Debug>(
    method: &str,
    params: serde_json::Value,
) -> Result<JsonRpcResult<T>, JsonRpcError> {
    let url = "http://127.0.0.1:18332".parse::<hyper::Uri>().unwrap();
    let host = url.host().expect("uri has no host");
    let port = url.port_u16().unwrap_or(18332);
    let address = format!("{}:{}", host, port);
    let stream = TcpStream::connect(address).await.unwrap();
    let io = TokioIo::new(stream);
    let (mut sender, conn) = hyper::client::conn::http1::handshake(io).await.unwrap();
    
    // 新增：启动任务处理连接
    tokio::spawn(async move {
        if let Err(e) = conn.await {
            eprintln!("Connection error: {}", e);
        }
    });
    
    let (username, password) = ("username", "password");
    let request = JsonRpcRequest {
        jsonrpc: "2.0".to_string(),
        method: method.to_string(),
        params,
        id: 1,
    };

    let request_body = serde_json::to_string(&request).unwrap();

    let req = Request::builder()
        .method("POST")
        .uri(url)
        .header(CONTENT_TYPE, "application/json")
        .header("Host", host)
        .header(
            AUTHORIZATION,
            format!(
                "Basic {}",
                base64::engine::general_purpose::STANDARD
                    .encode(format!("{}:{}", username, password))
            ),
        )
        .body(Full::<Bytes>::from(request_body))
        .unwrap();

    println!("A");
    let response = sender.send_request(req).await.unwrap();

    println!("B");
    let status = response.status();
    let body = response.into_body().collect().await.unwrap().to_bytes();

    if status.is_success() {
        serde_json::from_slice(&body).unwrap()
    } else {
        match serde_json::from_slice(&body) {
            Ok(error_response) => Err(error_response),
            Err(e) => Err(JsonRpcError {
                code: -1,
                message: format!("Deserialization error {:?}", e),
            }),
        }
    }
}

额外建议 #

• 替换代码中的unwrap()为?或显式错误处理，这样能更清晰地定位连接、请求构建等环节的问题。
• 查看比特币节点日志，确认是否收到请求以及是否有错误输出，辅助排查问题。

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