mysql_async

基于 Tokio 的 Rust 编程语言的异步 MySql 客户端库。

安装

该库托管在 crates.io 上。

[dependencies]
mysql_async = "<desired version>"

crate 特性

默认特性集较广 - 包括所有默认的 mysql_common 特性 以及基于 native-tls 的 TLS 支持。

特性列表

• minimal – 启用必要的特性（目前唯一的必要特性是 flate2 后端）。启用

  • `flate2/zlib"

  示例

  [dependencies]
  mysql_async = { version = "*", default-features = false, features = ["minimal"]}
  

  *注意：可以通过直接选择它来使用另一个 flate2 后端

  [dependencies]
  mysql_async = { version = "*", default-features = false }
  flate2 = { version = "*", default-features = false, features = ["rust_backend"] }
  

• default – 启用以下 crate 和依赖项的特性

  • native-tls-tls
  • `flate2/zlib"
  • mysql_common/bigdecimal03
  • mysql_common/rust_decimal
  • mysql_common/time03
  • mysql_common/uuid
  • mysql_common/frunk

• default-rustls – 与默认设置相同，但使用 rustls-tls 而不是 native-tls-tls。

  示例

  [dependencies]
  mysql_async = { version = "*", default-features = false, features = ["default-rustls"] }
  

• native-tls-tls – 启用基于 native-tls 的 TLS 支持 （与 rustls-tls 冲突）

  示例

  [dependencies]
  mysql_async = { version = "*", default-features = false, features = ["native-tls-tls"] }
  
  

• rustls-tls – 启用基于 native-tls 的 TLS 支持 （与 native-tls-tls 冲突）

  示例

  [dependencies]
  mysql_async = { version = "*", default-features = false, features = ["rustls-tls"] }
  
  

• tracing – 启用通过 tracing 包进行仪表化。

  主要操作（query、prepare、exec）在INFO级别进行了监控。其余操作，包括get_conn，在DEBUG级别进行监控。在DEBUG级别，还将SQL查询和参数添加到query、prepare和exec跨度中。还有一些内部查询在TRACE级别进行了监控。

  示例

  [dependencies]
  mysql_async = { version = "*", features = ["tracing"] }
  

• derive – 启用mysql_common/derive功能

TLS/SSL支持

SSL支持有两种形式

1. 基于native-tls – 这是默认选项，通常无需特别处理（请参阅native-tls-tls crate功能）。

2. 基于rustls – Rust编写的TLS后端（请参阅rustls-tls crate功能）。

   请注意关于rustls的几点

   • 如果您尝试通过IP地址连接到服务器，它将失败，需要主机名；
   • 它可能在Windows上不起作用，至少在使用MySql安装程序生成的默认服务器证书的情况下。

连接URL参数

驱动程序支持一组URL参数（请参阅Opts上的文档）。

示例

use mysql_async::prelude::*;

#[derive(Debug, PartialEq, Eq, Clone)]
struct Payment {
    customer_id: i32,
    amount: i32,
    account_name: Option<String>,
}

#[tokio::main]
async fn main() -> Result<()> {
    let payments = vec![
        Payment { customer_id: 1, amount: 2, account_name: None },
        Payment { customer_id: 3, amount: 4, account_name: Some("foo".into()) },
        Payment { customer_id: 5, amount: 6, account_name: None },
        Payment { customer_id: 7, amount: 8, account_name: None },
        Payment { customer_id: 9, amount: 10, account_name: Some("bar".into()) },
    ];

    let database_url = /* ... */
    # get_opts();

    let pool = mysql_async::Pool::new(database_url);
    let mut conn = pool.get_conn().await?;

    // Create a temporary table
    r"CREATE TEMPORARY TABLE payment (
        customer_id int not null,
        amount int not null,
        account_name text
    )".ignore(&mut conn).await?;

    // Save payments
    r"INSERT INTO payment (customer_id, amount, account_name)
      VALUES (:customer_id, :amount, :account_name)"
        .with(payments.iter().map(|payment| params! {
            "customer_id" => payment.customer_id,
            "amount" => payment.amount,
            "account_name" => payment.account_name.as_ref(),
        }))
        .batch(&mut conn)
        .await?;

    // Load payments from the database. Type inference will work here.
    let loaded_payments = "SELECT customer_id, amount, account_name FROM payment"
        .with(())
        .map(&mut conn, |(customer_id, amount, account_name)| Payment { customer_id, amount, account_name })
        .await?;

    // Dropped connection will go to the pool
    drop(conn);

    // The Pool must be disconnected explicitly because
    // it's an asynchronous operation.
    pool.disconnect().await?;

    assert_eq!(loaded_payments, payments);

    // the async fn returns Result, so
    Ok(())
}

池

Pool结构是一个异步连接池。

请注意

• Pool是一个智能指针 – 每个克隆都将指向同一个池实例。
• Pool是Send + Sync + 'static – 可以自由传递。
• 使用Pool::disconnect优雅地关闭池。
• ⚠️ Pool::new是懒加载的，不会断言服务器可用性。

事务

Conn::start_transaction是一个包装器，以START TRANSACTION开始，以COMMIT或ROLLBACK结束。

#
let pool = mysql_async::Pool::new(database_url);

let mut conn = pool.get_conn().await?;
"CREATE TEMPORARY TABLE tmp (id INT, val TEXT)".ignore(&mut conn).await?;

// We are going to call `LOAD DATA LOCAL` so let's setup a one-time handler.
conn.set_infile_handler(async move {
    // We need to return a stream of `io::Result<Bytes>`
    Ok(stream::iter([Bytes::from("1,a\r\n"), Bytes::from("2,b\r\n3,c")]).map(Ok).boxed())
});

let result = r#"LOAD DATA LOCAL INFILE 'whatever'
    INTO TABLE `tmp`
    FIELDS TERMINATED BY ',' ENCLOSED BY '\"'
    LINES TERMINATED BY '\r\n'"#.ignore(&mut conn).await;

match result {
    Ok(()) => (),
    Err(Error::Server(ref err)) if err.code == 1148 => {
        // The used command is not allowed with this MySQL version
        return Ok(());
    },
    Err(Error::Server(ref err)) if err.code == 3948 => {
        // Loading local data is disabled;
        // this must be enabled on both the client and the server
        return Ok(());
    }
    e @ Err(_) => e.unwrap(),
}

// Now let's verify the result
let result: Vec<(u32, String)> = conn.query("SELECT * FROM tmp ORDER BY id ASC").await?;
assert_eq!(
    result,
    vec![(1, "a".into()), (2, "b".into()), (3, "c".into())]
);

drop(conn);
pool.disconnect().await?;

docker run -d --name container \
    -v `pwd`:/root \
    -p 3307:3306 \
    -e MYSQL_ROOT_PASSWORD=password \
    mysql:8.0 \
    --max-allowed-packet=36700160 \
    --local-infile \
    --log-bin=mysql-bin \
    --log-slave-updates \
    --gtid_mode=ON \
    --enforce_gtid_consistency=ON \
    --server-id=1