PHP小码农

Appearance

MongoDB PHP 扩展 ​

概述 ​

MongoDB PHP 扩展（ext-mongodb）是 PHP 与 MongoDB 交互的低级别驱动。官方还提供了高层封装库 mongodb/mongodb（通过 Composer 安装），提供了更友好的 API。生产环境推荐使用 Composer 库。

两个层次的选择

层次包名安装方式API 风格
驱动层ext-mongodbPECL/编译MongoDB\Driver\*
库层mongodb/mongodbComposerMongoDB\Client

基础概念 ​

安装与配置 ​

bash
# 安装驱动扩展
pecl install mongodb

# Docker 安装
docker-php-ext-install mongodb

# 验证安装
php -m | grep mongodb

# 安装 Composer 库
composer require mongodb/mongodb

# 查看版本
php -r 'echo MongoDB\Driver\Server::VERSION;'

核心类层次 ​

类说明用途
MongoDB\Driver\Manager连接管理器创建连接、执行命令
MongoDB\Driver\Client客户端封装CRUD 操作（Composer 库）
MongoDB\Driver\Database数据库对象数据库级操作
MongoDB\Driver\Collection集合对象集合级 CRUD
MongoDB\Driver\BulkWrite批量写入批量 insert/update/delete
MongoDB\Driver\Query查询对象find 查询封装
MongoDB\Driver\Command命令对象执行数据库命令

语法与代码 ​

连接字符串 URI ​

php
<?php
declare(strict_types=1);

// 标准 MongoDB URI
$uri = 'mongodb://localhost:27017';

// 带认证
$uri = 'mongodb://username:password@localhost:27017';

// 指定数据库和选项
$uri = 'mongodb://user:pass@host1:27017,host2:27017/admin?replicaSet=myReplica&ssl=true';

// URI 选项说明:
// replicaSet — 副本集名称
// authSource — 认证数据库（默认 admin）
// ssl / tls — 启用 TLS
// connectTimeoutMS — 连接超时（毫秒）
// socketTimeoutMS — Socket 超时
// serverSelectionTimeoutMS — 服务器选择超时
// w — 写关注级别（1/0/majority）
// readPreference — 读取偏好（primary/secondary/nearest）
// retryWrites — 自动重试写入（true）
// retryReads — 自动重试读取（true）

使用驱动层 API ​

php
<?php
declare(strict_types=1);

use MongoDB\Driver\Manager;
use MongoDB\Driver\BulkWrite;
use MongoDB\Driver\Query;
use MongoDB\Driver\Command;

// 创建 Manager（连接池，延迟连接）
$manager = new Manager('mongodb://localhost:27017', [
    'connectTimeoutMS' => 5000,
    'socketTimeoutMS'  => 30000,
    'retryWrites'      => true,
    'retryReads'       => true,
    'w'                => 'majority',    // 写关注
    'readPreference'   => 'primary',     // 读偏好
]);

// 插入文档
$bulk = new BulkWrite();
$bulk->insert(['name' => '张三', 'age' => 28, 'email' => 'zhangsan@example.com']);
$bulk->insert(['name' => '李四', 'age' => 32, 'email' => 'lisi@example.com']);

$result = $manager->executeBulkWrite('testdb.users', $bulk);
echo "插入数: " . $result->getInsertedCount() . "\n";

// 查询文档
$query = new Query(['age' => ['$gte' => 30]], [
    'sort' => ['age' => 1],
    'limit' => 10,
]);

$cursor = $manager->executeQuery('testdb.users', $query);

foreach ($cursor as $document) {
    var_dump((array) $document);
}

// 执行命令
$command = new Command(['ping' => 1]);
$result = $manager->executeCommand('testdb', $command);
var_dump($result->toArray());

使用 Composer 库（推荐） ​

php
<?php
declare(strict_types=1);

use MongoDB\Client;

// 创建客户端
$client = new Client('mongodb://localhost:27017', [
    'connectTimeoutMS' => 5000,
    'retryWrites'     => true,
]);

// 选择数据库和集合
$database = $client->selectDatabase('testdb');
$collection = $client->selectCollection('testdb', 'users');

// 插入
$insertResult = $collection->insertOne([
    'name' => '张三',
    'age' => 28,
    'email' => 'zhangsan@example.com',
    'createdAt' => new MongoDB\BSON\UTCDateTime(),
]);
echo "插入ID: " . $insertResult->getInsertedId() . "\n";

// 查询
$user = $collection->findOne(['name' => '张三']);
if ($user) {
    echo "用户: " . $user['name'] . ", 年龄: " . $user['age'] . "\n";
}

BSON 类型

MongoDB 使用 BSON（Binary JSON）格式存储数据。PHP 驱动提供了 MongoDB\BSON\* 类来处理 BSON 类型和 PHP 类型的转换：

  • UTCDateTime — 日期时间
  • ObjectId — 文档 ID
  • Binary — 二进制数据
  • JavaScript — JavaScript 代码
  • Regex — 正则表达式

实战示例 ​

MongoDB 连接管理类 ​

php
<?php
declare(strict_types=1);

use MongoDB\Client;

class MongoDBFactory
{
    private static ?Client $instance = null;
    private static array $config = [];

    public static function configure(array $config): void
    {
        self::$config = array_merge([
            'uri'      => 'mongodb://localhost:27017',
            'database' => 'app_db',
            'options'  => [
                'connectTimeoutMS' => 5000,
                'socketTimeoutMS'  => 30000,
                'retryWrites'      => true,
                'retryReads'       => true,
                'w'                => 'majority',
                'readPreference'   => 'primary',
            ],
        ], $config);
    }

    public static function getClient(): Client
    {
        if (self::$instance === null) {
            self::$instance = new Client(
                self::$config['uri'],
                self::$config['options']
            );
        }
        return self::$instance;
    }

    public static function getCollection(string $collectionName): MongoDB\Driver\Collection
    {
        return self::getClient()->selectCollection(
            self::$config['database'],
            $collectionName
        );
    }

    public static function getDatabase(): MongoDB\Driver\Database
    {
        return self::getClient()->selectDatabase(self::$config['database']);
    }

    /**
     * 测试连接
     */
    public static function ping(): bool
    {
        try {
            $result = self::getClient()->selectDatabase('admin')->command(['ping' => 1]);
            return true;
        } catch (MongoDB\Driver\Exception\Exception $e) {
            error_log("MongoDB 连接失败: " . $e->getMessage());
            return false;
        }
    }
}

BulkWrite 批量操作 ​

php
<?php
declare(strict_types=1);

use MongoDB\Driver\Manager;
use MongoDB\Driver\BulkWrite;
use MongoDB\Driver\Exception\BulkWriteException;

class BulkOperations
{
    private Manager $manager;

    public function __construct(Manager $manager)
    {
        $this->manager = $manager;
    }

    /**
     * 批量混合操作（insert + update + delete）
     */
    public function mixedBulkWrite(string $namespace, array $operations): array
    {
        $bulk = new BulkWrite(['ordered' => false]); // 无序执行，并行

        foreach ($operations as $op) {
            switch ($op['type']) {
                case 'insert':
                    $bulk->insert($op['document']);
                    break;
                case 'update':
                    $bulk->update(
                        $op['filter'],
                        $op['update'],
                        $op['options'] ?? ['multi' => false, 'upsert' => false]
                    );
                    break;
                case 'delete':
                    $bulk->delete(
                        $op['filter'],
                        $op['options'] ?? ['limit' => 1]
                    );
                    break;
            }
        }

        try {
            $result = $this->manager->executeBulkWrite($namespace, $bulk);
            return [
                'insertedCount' => $result->getInsertedCount(),
                'matchedCount'  => $result->getMatchedCount(),
                'modifiedCount' => $result->getModifiedCount(),
                'deletedCount'  => $result->getDeletedCount(),
                'upsertedCount' => $result->getUpsertedCount(),
            ];
        } catch (BulkWriteException $e) {
            $writeResult = $e->getWriteResult();
            throw new RuntimeException(
                "批量写入失败: " . implode(', ', $writeResult->getWriteErrors()),
                0,
                $e
            );
        }
    }

    /**
     * 批量导入数据
     */
    public function importFromJson(string $namespace, string $filePath): int
    {
        $data = json_decode(file_get_contents($filePath), true);
        if (!is_array($data)) {
            throw new RuntimeException("JSON 文件格式错误");
        }

        $bulk = new BulkWrite(['ordered' => false]);
        foreach ($data as $document) {
            $bulk->insert($document);
        }

        $result = $this->manager->executeBulkWrite($namespace, $bulk);
        return $result->getInsertedCount();
    }
}

注意事项 ​

连接超时与重试 ​

php
<?php
// 连接选项详解
$options = [
    // 连接阶段超时
    'connectTimeoutMS' => 5000,     // 5秒连接超时

    // Socket 操作超时
    'socketTimeoutMS'  => 0,        // 0 = 无超时（不推荐）

    // 服务器选择超时
    'serverSelectionTimeoutMS' => 30000, // 30秒

    // 心跳间隔
    'heartbeatFrequencyMS' => 10000, // 10秒

    // 自动重试
    'retryWrites' => true,   // 自动重试可重试的写入操作
    'retryReads'  => true,   // 自动重试可重试的读取操作

    // 连接池设置
    'maxPoolSize' => 100,    // 最大连接数
    'minPoolSize' => 0,      // 最小连接数
    'maxIdleTimeMS' => 120000, // 空闲连接超时 2分钟
];

BSON 类型转换 ​

php
<?php
use MongoDB\BSON\UTCDateTime;
use MongoDB\BSON\ObjectId;
use MongoDB\BSON\Binary;

// PHP DateTime → BSON UTCDateTime
$phpDate = new DateTime('2024-01-15 10:30:00');
$bsonDate = new UTCDateTime($phpDate->getTimestamp() * 1000);

// BSON UTCDateTime → PHP DateTime
$phpDate = $bsonDate->toDateTime()->setTimezone(new DateTimeZone('Asia/Shanghai'));

// 创建 ObjectId
$objectId = new ObjectId();
$stringId = (string) $objectId; // 十六进制字符串

// 从字符串创建 ObjectId
$objectId = new ObjectId('65a1b2c3d4e5f6a7b8c9d0e1');

// Binary 数据（存储文件、图片等）
$binary = new Binary(file_get_contents('image.jpg'), Binary::TYPE_GENERIC);

时区问题

UTCDateTime 总是以 UTC 存储。读取时需要手动转换为本地时区。建议在应用层统一使用 UTCDateTime 进行存储和查询。

最佳实践 ​

1. 连接管理 ​

php
<?php
// 推荐: 全局单例 Client
// Client 内部维护连接池，不需要手动管理连接
// Manager 和 Client 都是线程安全的（在 PHP-FPM 中每个进程独立）

// 不推荐: 每次请求创建新 Client
// 原因: 虽然连接池会复用，但创建对象有开销

2. 环境隔离 ​

php
<?php
// 不同环境使用不同的数据库
$env = getenv('APP_ENV') ?: 'development';

$configs = [
    'development' => [
        'uri' => 'mongodb://localhost:27017',
        'database' => 'app_dev',
    ],
    'testing' => [
        'uri' => 'mongodb://localhost:27017',
        'database' => 'app_test',
    ],
    'production' => [
        'uri' => 'mongodb://user:pass@mongo1:27017,mongo2:27017/admin?replicaSet=rs0',
        'database' => 'app_prod',
        'options' => [
            'ssl' => true,
            'w' => 'majority',
            'readPreference' => 'secondaryPreferred',
        ],
    ],
];

MongoDBFactory::configure($configs[$env]);

参考链接 ​