MySQLi 常用操作

概述

MySQLi 扩展提供了丰富的数据库操作方法，包括单条查询、多语句查询、预处理语句和结果集处理。理解不同查询方式的适用场景是编写高效 MySQL 操作代码的基础。

方法选择

• query() — 执行单条 SQL，返回结果集或布尔值
• real_query() — 执行 SQL 但不获取/缓冲结果集
• multi_query() — 执行多条 SQL 语句
• prepare() — 预处理语句（安全推荐）

基础概念

查询方法对比

方法	返回值	缓冲	适用场景
query()	mysqli_result	是	SELECT/SHOW
real_query()	bool	否	大结果集
multi_query()	bool	是	多语句/批量
prepare()	mysqli_stmt	否	参数化查询

结果处理方式

方法	返回类型	说明
fetch_row()	索引数组	['张三', 28, 'z@e.com']
fetch_assoc()	关联数组	['name' => '张三', 'age' => 28]
fetch_array()	两者都有	同时支持索引和键名
fetch_object()	对象	$row->name
fetch_all()	二维数组	获取全部结果（MySQLnd）

语法与代码

基本查询与结果处理

<?php
declare(strict_types=1);

$mysqli = new mysqli('localhost', 'user', 'pass', 'app_db');
$mysqli->set_charset('utf8mb4');

// SELECT 查询
$result = $mysqli->query('SELECT id, name, email, age FROM users WHERE status = "active"');

if ($result === false) {
    die("查询失败: " . $mysqli->error);
}

// fetch_assoc — 关联数组
while ($row = $result->fetch_assoc()) {
    printf(
        "ID: %d | 姓名: %s | 邮箱: %s | 年龄: %d\n",
        $row['id'],
        $row['name'],
        $row['email'],
        $row['age']
    );
}

// fetch_row — 索引数组
$result->data_seek(0); // 重置指针
while ($row = $result->fetch_row()) {
    echo "ID: {$row[0]}, 姓名: {$row[1]}\n";
}

// fetch_object — 对象风格
$result->data_seek(0);
while ($row = $result->fetch_object()) {
    echo "ID: {$row->id}, 姓名: {$row->name}\n";
}

// fetch_all — 一次性获取全部（MySQLnd）
$users = $result->fetch_all(MYSQLI_ASSOC);

// 释放结果集
$result->free();

// 获取行数和字段数
echo "总行数: " . $result->num_rows . "\n";
echo "字段数: " . $result->field_count . "\n";

写入操作

<?php
declare(strict_types=1);

$mysqli = new mysqli('localhost', 'user', 'pass', 'app_db');

// INSERT
$mysqli->query("
    INSERT INTO users (name, email, age, status)
    VALUES ('张三', 'zhangsan@example.com', 28, 'active')
");

echo "插入ID: " . $mysqli->insert_id . "\n";
echo "影响行数: " . $mysqli->affected_rows . "\n";

// UPDATE
$mysqli->query("UPDATE users SET age = 29 WHERE id = " . $mysqli->insert_id);
echo "更新行数: " . $mysqli->affected_rows . "\n";

// DELETE
$mysqli->query("DELETE FROM users WHERE id = 999");
echo "删除行数: " . $mysqli->affected_rows . "\n";

// REPLACE — 不存在则插入，存在则先删除再插入
$mysqli->query("
    REPLACE INTO config (key, value)
    VALUES ('site_name', '我的网站')
");

multi_query — 多语句查询

<?php
declare(strict_types=1);

$mysqli = new mysqli('localhost', 'user', 'pass', 'app_db');

// 执行多条 SQL
$sql = "
    SELECT COUNT(*) as total FROM users;
    SELECT COUNT(*) as active FROM users WHERE status = 'active';
    SELECT COUNT(*) as banned FROM users WHERE status = 'banned';
";

if ($mysqli->multi_query($sql)) {
    do {
        $result = $mysqli->store_result();
        if ($result) {
            $row = $result->fetch_assoc();
            echo "结果: " . json_encode($row) . "\n";
            $result->free();
        }
    } while ($mysqli->more_results() && $mysqli->next_result());
}

if ($mysqli->error) {
    echo "错误: " . $mysqli->error . "\n";
}

// multi_query 批量插入（不推荐，优先使用预处理批量）
$insertSql = '';
foreach ($data as $row) {
    $insertSql .= "INSERT INTO logs (message) VALUES ('" . $mysqli->real_escape_string($row['msg']) . "');";
}
$mysqli->multi_query($insertSql);

multi_query 安全风险

multi_query 不能使用预处理语句，因此无法自动防护 SQL 注入。使用时必须手动转义所有输入值。推荐在非用户输入场景（如批量 DDL 操作）中使用。

affected_rows 与 insert_id

<?php
declare(strict_types=1);

$mysqli = new mysqli('localhost', 'user', 'pass', 'app_db');

// INSERT — insert_id 返回最后插入的 ID
$mysqli->query("INSERT INTO users (name) VALUES ('test1')");
$firstId = $mysqli->insert_id; // 1

$mysqli->query("INSERT INTO users (name) VALUES ('test2')");
$secondId = $mysqli->insert_id; // 2

// 批量插入 — insert_id 返回第一个 ID
$mysqli->query("
    INSERT INTO users (name) VALUES ('a'), ('b'), ('c')
");
$firstBatchId = $mysqli->insert_id; // 3 (a=3, b=4, c=5)

// UPDATE — affected_rows 返回修改的行数
$mysqli->query("UPDATE users SET status = 'inactive' WHERE status = 'active' AND age < 18");
$updated = $mysqli->affected_rows;

// REPLACE — affected_rows 包含删除的行
$mysqli->query("REPLACE INTO config (key, value) VALUES ('name', 'new_value')");
// affected_rows: 2（1删 + 1插）或 1（直接插入）

// 注意: 如果 UPDATE 的值与原值相同，affected_rows 为 0
$mysqli->query("UPDATE users SET status = 'active' WHERE status = 'active'");
echo "影响行数: " . $mysqli->affected_rows; // 0

real_query — 非缓冲查询

<?php
declare(strict_types=1);

// real_query 不缓冲结果集 — 适合大数据量查询
$mysqli->real_query("SELECT * FROM large_table");

// 使用 use_result() 获取非缓冲结果集
$result = $mysqli->use_result();

while ($row = $result->fetch_assoc()) {
    // 逐行读取，不一次性加载到内存
    // 但注意: 调用期间不能执行其他查询
}

$result->free();

// 对比: query() 使用 store_result()
$result = $mysqli->query("SELECT * FROM large_table");
// 等同于
$mysqli->real_query("SELECT * FROM large_table");
$result = $mysqli->store_result(); // 全部缓冲到内存

实战示例

MySQLi 辅助类

<?php
declare(strict_types=1);

class MySQLiHelper
{
    private mysqli $mysqli;

    public function __construct(mysqli $mysqli)
    {
        $this->mysqli = $mysqli;
        $this->mysqli->set_charset('utf8mb4');
    }

    /**
     * 查询单行
     */
    public function fetchOne(string $sql, array $params = []): ?array
    {
        $stmt = $this->mysqli->prepare($sql);
        if (!$stmt) {
            throw new RuntimeException($this->mysqli->error);
        }

        if (!empty($params)) {
            $types = $this->getParamTypes($params);
            $stmt->bind_param($types, ...$params);
        }

        $stmt->execute();
        $result = $stmt->get_result();
        $row = $result->fetch_assoc();
        $stmt->close();

        return $row ?: null;
    }

    /**
     * 查询多行
     */
    public function fetchAll(string $sql, array $params = []): array
    {
        $stmt = $this->mysqli->prepare($sql);
        if (!$stmt) {
            throw new RuntimeException($this->mysqli->error);
        }

        if (!empty($params)) {
            $types = $this->getParamTypes($params);
            $stmt->bind_param($types, ...$params);
        }

        $stmt->execute();
        $result = $stmt->get_result();
        $rows = $result->fetch_all(MYSQLI_ASSOC);
        $stmt->close();

        return $rows;
    }

    /**
     * 分页查询
     */
    public function paginate(string $sql, array $params, int $page, int $perPage): array
    {
        $countSql = preg_replace('/SELECT .*? FROM/i', 'SELECT COUNT(*) as total FROM', $sql);
        $countSql = preg_replace('/ORDER BY.*$/i', '', $countSql);
        $countSql = preg_replace('/LIMIT.*$/i', '', $countSql);

        $total = $this->fetchOne($countSql, $params)['total'] ?? 0;

        $pageSql = $sql . " LIMIT " . (($page - 1) * $perPage) . ", {$perPage}";
        $rows = $this->fetchAll($pageSql, $params);

        return [
            'data' => $rows,
            'total' => (int) $total,
            'page' => $page,
            'perPage' => $perPage,
            'totalPages' => (int) ceil($total / $perPage),
        ];
    }

    /**
     * 执行写入
     */
    public function execute(string $sql, array $params = []): int
    {
        $stmt = $this->mysqli->prepare($sql);
        if (!$stmt) {
            throw new RuntimeException($this->mysqli->error);
        }

        if (!empty($params)) {
            $types = $this->getParamTypes($params);
            $stmt->bind_param($types, ...$params);
        }

        $stmt->execute();
        $affected = $stmt->affected_rows;
        $stmt->close();

        return $affected;
    }

    /**
     * 插入并返回 ID
     */
    public function insert(string $sql, array $params = []): int
    {
        $this->execute($sql, $params);
        return $this->mysqli->insert_id;
    }

    private function getParamTypes(array $params): string
    {
        $types = '';
        foreach ($params as $param) {
            $types .= match (true) {
                is_int($param) => 'i',
                is_float($param) => 'd',
                is_null($param) => 'b',
                default => 's',
            };
        }
        return $types;
    }
}

结果集处理工具

<?php
declare(strict_types=1);

class ResultSetHelper
{
    /**
     * 获取单列值数组
     */
    public static function pluck(mysqli_result $result, string $column): array
    {
        $values = [];
        while ($row = $result->fetch_assoc()) {
            $values[] = $row[$column];
        }
        return $values;
    }

    /**
     * 以某列值为键的关联数组
     */
    public static function keyBy(mysqli_result $result, string $keyColumn): array
    {
        $map = [];
        while ($row = $result->fetch_assoc()) {
            $map[$row[$keyColumn]] = $row;
        }
        return $map;
    }

    /**
     * JSON 输出
     */
    public static function toJson(mysqli_result $result): string
    {
        return json_encode($result->fetch_all(MYSQLI_ASSOC), JSON_UNESCAPED_UNICODE);
    }

    /**
     * CSV 导出
     */
    public static function toCsv(mysqli_result $result, string $filePath): int
    {
        $handle = fopen($filePath, 'w');
        if ($handle === false) {
            throw new RuntimeException("无法创建 CSV 文件");
        }

        // 写入标题
        $fields = $result->fetch_fields();
        $headers = array_map(fn($f) => $f->name, $fields);
        fputcsv($handle, $headers);

        // 写入数据
        $count = 0;
        while ($row = $result->fetch_assoc()) {
            fputcsv($handle, $row);
            $count++;
        }

        fclose($handle);
        return $count;
    }
}

注意事项

错误处理

<?php
// MySQLi 错误处理最佳实践

// 方式1: 异常模式（推荐）
mysqli_report(MYSQLI_REPORT_ERROR | MYSQLI_REPORT_STRICT);
try {
    $mysqli = new mysqli('localhost', 'user', 'pass', 'app_db');
    $mysqli->query("SELECT * FROM nonexistent_table");
} catch (mysqli_sql_exception $e) {
    error_log("MySQL 错误 [{$e->getCode()}]: {$e->getMessage()}");
    echo "数据库操作失败\n";
}

// 方式2: 手动检查
if ($mysqli->connect_errno) {
    die("连接失败: " . $mysqli->connect_error);
}

if (!$result = $mysqli->query($sql)) {
    error_log("查询失败: " . $mysqli->error);
}

// 注意: 不要将 $mysqli->error 直接输出给用户
// 可能暴露数据库结构信息

内存管理

<?php
// 大结果集处理

// 方式1: 非缓冲查询（real_query + use_result）
$mysqli->real_query("SELECT * FROM million_rows_table");
$result = $mysqli->use_result();
while ($row = $result->fetch_assoc()) {
    // 逐行处理，内存占用低
    // 但此期间连接被锁定，不能执行其他查询
}
$result->free();

// 方式2: 分批查询
$pageSize = 10000;
$offset = 0;
while (true) {
    $result = $mysqli->query("SELECT * FROM large_table LIMIT {$pageSize} OFFSET {$offset}");
    $rows = $result->fetch_all(MYSQLI_ASSOC);
    $result->free();

    if (empty($rows)) break;

    foreach ($rows as $row) {
        // 处理数据
    }

    $offset += $pageSize;
}

// 方式3: 使用游标分页（推荐大数据量）
$lastId = 0;
$pageSize = 1000;
while (true) {
    $result = $mysqli->query("SELECT * FROM large_table WHERE id > {$lastId} ORDER BY id LIMIT {$pageSize}");
    $rows = $result->fetch_all(MYSQLI_ASSOC);
    $result->free();

    if (empty($rows)) break;

    foreach ($rows as $row) {
        // 处理
    }

    $lastId = $row['id'] ?? 0;
}

最佳实践

1. 编码设置

<?php
// 必须设置字符集为 utf8mb4
$mysqli = new mysqli('localhost', 'user', 'pass', 'app_db');
$mysqli->set_charset('utf8mb4'); // 推荐
// 或
$mysqli->query("SET NAMES utf8mb4");

// 不要使用 utf8（MySQL 的 utf8 只支持 3 字节，不是真正的 UTF-8）
// utf8mb4 支持 4 字节，包括 emoji 字符

2. 连接选项

<?php
$mysqli = new mysqli();
$mysqli->options(MYSQLI_OPT_CONNECT_TIMEOUT, 5);    // 连接超时5秒
$mysqli->options(MYSQLI_OPT_READ_TIMEOUT, 30);       // 读取超时30秒
$mysqli->options(MYSQLI_OPT_INT_AND_FLOAT_NATIVE, true); // 原生类型返回
$mysqli->real_connect('localhost', 'user', 'pass', 'app_db');

参考链接

• PHP 手册: mysqli
• PHP 手册: mysqli_result
• MySQL 字符集和排序规则
• PHP mysqli_result::fetch_all