mysql_async

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

安装

该库托管在 crates.io。

[dependencies]
mysql_async = "<desired version>"

包功能

默认功能集很广 - 它包括所有默认的 mysql_common 功能 以及基于 native-tls 的 TLS 支持。

功能列表

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

  • `flate2/zlib"

  示例

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

• minimal-rust - 与 minimal 相同，但选择基于 rust 的 flate2 后端。启用

  • flate2/rust_backend

• default – 启用以下功能集

  • minimal
  • native-tls-tls
  • bigdecimal
  • rust_decimal
  • time
  • frunk
  • binlog

• 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"] }
  

• binlog - 启用binlog相关功能。启用

  • `mysql_common/binlog"

代理功能

• derive – 启用mysql_common/derive功能
• chrono = 启用mysql_common/chrono功能
• time = 启用mysql_common/time功能
• bigdecimal = 启用mysql_common/bigdecimal功能
• rust_decimal = 启用mysql_common/rust_decimal功能
• frunk = 启用mysql_common/frunk功能

TLS/SSL 支持

SSL支持有两种形式

1. 基于native-tls – 这是默认选项，通常无需担心（参见native-tls-tls包功能）。

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

   请注意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是一个智能指针 – 每个clone都将指向同一个池实例。
• Pool是Send + Sync + 'static – 可以自由传递它。
• 使用Pool::disconnect来优雅地关闭池。
• ⚠️ Pool::new是懒惰的，不会断言服务器可用性。

事务

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

简单来说，全局处理器是一个异步函数，它接受一个文件名（作为 &[u8]）并返回 Result<InfileData>。

您可以使用 OptsBuilder::local_infile_handler 来设置它。如果没有为连接安装本地处理器，则服务器将使用它。此处理器可能会被调用多次。

示例

1. WhiteListFsHandler 是一个 全局 处理器。
2. 每个 T: Fn(&[u8]) -> BoxFuture<'static, Result<InfileData, LocalInfileError>> 都是 全局 处理器。

本地 本地 INFILE 处理器。

简单来说，本地处理器是一个返回 Result<InfileData> 的未来。

这是一个一次性处理器 - 使用后会被消耗。您可以使用 Conn::set_infile_handler 来设置它。此处理器优先于 全局 处理器。

值得注意

1. impl Drop for Conn 将清除本地处理器，即处理器将在连接返回到 Pool 时被移除。
2. Conn::reset 将清除本地处理器。

示例

#
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?;

测试

测试使用了以下环境变量

• DATABASE_URL - 默认为 mysql://root:[email protected]:3307/mysql
• COMPRESS - 设置为 1 或 true 以启用测试的压缩
• SSL - 设置为 1 或 true 以启用测试的 TLS

您可以使用 Docker 运行测试服务器。请注意，有关最大允许数据包、本地 INFILE 和二进制日志的参数对于正确运行测试是必需的（请参阅 azure-pipelines.yml）

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

变更日志

可在 此处 查看

许可证

许可协议为以下之一

• Apache License, Version 2.0, (LICENSE-APACHE 或 https://www.apache.org/licenses/LICENSE-2.0)
• 麻省理工学院许可证（LICENSE-MIT 或 https://opensource.org/licenses/MIT）

任选其一。

贡献

除非您明确表示，否则根据Apache-2.0许可证定义，您有意提交给作品的所有贡献，应按上述方式双许可，不附加任何额外条款或条件。