PHP `CQRS` (Command Query Responsibility Segregation) 架构在大型应用中的实践

各位听众，大家好！我是你们的老朋友，今天咱们聊聊PHP领域里一个听起来高大上，用起来真香的架构模式——CQRS，也就是命令查询职责分离。这玩意儿，说白了，就是把读和写操作彻底分开，让你的大型应用跑得更快，维护起来更轻松。别怕，今天咱用大白话、结合代码，把这玩意儿彻底讲透。

开场白：为啥要搞CQRS？

想象一下，你的电商网站，每天几百万用户访问，商品信息需要频繁更新，订单数据爆炸式增长。如果读写操作都挤在一个数据库，一个表里，那会发生什么？

性能瓶颈： 读写互相干扰，用户访问卡顿，下单失败，老板脸色难看。
扩展困难： 数据库压力巨大，想加机器扩容，发现牵一发而动全身。
代码复杂： 一个接口既要处理读取逻辑，又要处理写入逻辑，代码臃肿，难以维护，改个bug，可能引入更多bug。

CQRS就是为了解决这些问题而生的。它把读写操作拆分成两个独立的部分，分别使用不同的数据模型和处理方式，从而提高性能、可扩展性和可维护性。

CQRS的核心思想

CQRS的核心思想可以用一句话概括：读写分离，术业有专攻。

Command（命令）： 用于修改系统状态的操作，比如创建用户、更新商品信息、下单等。Command通常不返回任何数据，或者只返回操作是否成功的状态。
Query（查询）： 用于读取系统状态的操作，比如获取用户信息、查询商品列表、获取订单详情等。Query通常返回需要的数据。

CQRS的组成部分

一个典型的CQRS架构通常包含以下几个部分：

Commands（命令）： 代表用户的意图，例如“创建用户”、“更新商品”。
Command Handlers（命令处理器）： 接收命令，执行相应的业务逻辑，并修改系统状态。
Events（事件）： 表示系统中发生的事件，例如“用户已创建”、“商品已更新”。
Event Store（事件存储）： 持久化存储所有事件，用于审计、回溯和重建状态。
Queries（查询）： 用于读取系统状态的请求，例如“获取用户列表”、“查询商品详情”。
Query Handlers（查询处理器）： 接收查询，从读取数据模型中获取数据，并返回结果。
Read Database（读取数据库）： 专门用于读取数据的数据库，通常针对读取优化，例如使用缓存、索引等。
Write Database（写入数据库）： 专门用于写入数据的数据库，通常针对写入优化，例如使用消息队列、批量写入等。

CQRS的优势与劣势

优势：

性能提升： 读写分离，可以针对读写操作分别进行优化，提高系统整体性能。
可扩展性： 读写分离，可以独立扩展读写数据库，满足不同业务需求。
可维护性： 代码职责清晰，易于理解和维护。
灵活性： 可以使用不同的数据模型和技术栈处理读写操作，提高系统灵活性。

劣势：

复杂性增加： 引入了更多的组件和概念，增加了系统的复杂性。
最终一致性： 读写分离可能导致数据不一致，需要处理最终一致性问题。
学习成本： 需要学习新的架构模式和技术栈。

PHP中的CQRS实践：以用户管理为例

咱们用一个简单的用户管理系统为例，演示如何在PHP中实践CQRS。

1. 定义Commands和Queries

首先，定义一些Command和Query类，用于表示用户的意图和查询请求。

<?php

// Commands
interface Command {}

class CreateUserCommand implements Command
{
    public string $username;
    public string $email;

    public function __construct(string $username, string $email)
    {
        $this->username = $username;
        $this->email = $email;
    }
}

class UpdateUserEmailCommand implements Command
{
    public int $userId;
    public string $email;

    public function __construct(int $userId, string $email)
    {
        $this->userId = $userId;
        $this->email = $email;
    }
}

// Queries
interface Query {}

class GetUserByIdQuery implements Query
{
    public int $userId;

    public function __construct(int $userId)
    {
        $this->userId = $userId;
    }
}

class GetAllUsersQuery implements Query {}

2. 定义Command Handlers和Query Handlers

接下来，定义Command Handler和Query Handler，用于处理命令和查询请求。

<?php

interface CommandHandler
{
    public function handle(Command $command);
}

interface QueryHandler
{
    public function handle(Query $query);
}

class CreateUserCommandHandler implements CommandHandler
{
    private WriteUserRepository $userRepository;

    public function __construct(WriteUserRepository $userRepository)
    {
        $this->userRepository = $userRepository;
    }

    public function handle(Command $command)
    {
        if (!$command instanceof CreateUserCommand) {
            throw new InvalidArgumentException('Invalid command type.');
        }

        $user = new User($command->username, $command->email);
        $this->userRepository->save($user);

        // 触发一个 UserCreatedEvent (可选)
        $this->publishEvent(new UserCreatedEvent($user->getId(), $user->getUsername(), $user->getEmail()));
    }

     private function publishEvent(UserCreatedEvent $event): void
    {
        // 在这里将事件发布到事件总线或消息队列
        // 例如：$this->eventBus->publish($event);
        // 或者：$this->messageQueue->enqueue($event);

        // 为了简单起见，这里我们只做个简单的日志记录
        error_log("UserCreatedEvent: User ID={$event->getUserId()}, Username={$event->getUsername()}, Email={$event->getEmail()}");
    }
}

class UpdateUserEmailCommandHandler implements CommandHandler
{
    private WriteUserRepository $userRepository;

    public function __construct(WriteUserRepository $userRepository)
    {
        $this->userRepository = $userRepository;
    }

    public function handle(Command $command)
    {
        if (!$command instanceof UpdateUserEmailCommand) {
            throw new InvalidArgumentException('Invalid command type.');
        }

        $user = $this->userRepository->find($command->userId);
        if (!$user) {
            throw new RuntimeException('User not found.');
        }

        $user->setEmail($command->email);
        $this->userRepository->save($user);

        // 触发一个 UserEmailUpdatedEvent (可选)
    }
}

class GetUserByIdQueryHandler implements QueryHandler
{
    private ReadUserRepository $userRepository;

    public function __construct(ReadUserRepository $userRepository)
    {
        $this->userRepository = $userRepository;
    }

    public function handle(Query $query)
    {
        if (!$query instanceof GetUserByIdQuery) {
            throw new InvalidArgumentException('Invalid query type.');
        }

        return $this->userRepository->find($query->userId);
    }
}

class GetAllUsersQueryHandler implements QueryHandler
{
    private ReadUserRepository $userRepository;

    public function __construct(ReadUserRepository $userRepository)
    {
        $this->userRepository = $userRepository;
    }

    public function handle(Query $query)
    {
        if (!$query instanceof GetAllUsersQuery) {
            throw new InvalidArgumentException('Invalid query type.');
        }

        return $this->userRepository->findAll();
    }
}

3. 定义Repositories

定义Repository接口和实现，用于访问数据库。注意，这里我们区分了Read Repository和Write Repository。

<?php

interface WriteUserRepository
{
    public function save(User $user): void;
    public function find(int $id): ?User;
}

interface ReadUserRepository
{
    public function find(int $id): ?User;
    public function findAll(): array;
}

// 使用 Doctrine ORM 的示例（仅作演示）
use DoctrineORMEntityManagerInterface;

class DoctrineWriteUserRepository implements WriteUserRepository
{
    private EntityManagerInterface $entityManager;

    public function __construct(EntityManagerInterface $entityManager)
    {
        $this->entityManager = $entityManager;
    }

    public function save(User $user): void
    {
        $this->entityManager->persist($user);
        $this->entityManager->flush();
    }

    public function find(int $id): ?User
    {
        return $this->entityManager->find(User::class, $id);
    }
}

class DoctrineReadUserRepository implements ReadUserRepository
{
    private EntityManagerInterface $entityManager;  // 这里也可以使用专门的 Read Only EntityManager

    public function __construct(EntityManagerInterface $entityManager)
    {
        $this->entityManager = $entityManager;
    }

    public function find(int $id): ?User
    {
        return $this->entityManager->find(UserReadModel::class, $id); // 注意这里使用的是 UserReadModel
    }

    public function findAll(): array
    {
        return $this->entityManager->getRepository(UserReadModel::class)->findAll();  // 注意这里使用的是 UserReadModel
    }
}

4. 定义Command Bus和Query Bus

Command Bus和Query Bus用于将命令和查询分发给相应的Handler。

<?php

interface CommandBus
{
    public function dispatch(Command $command);
}

interface QueryBus
{
    public function query(Query $query);
}

class SimpleCommandBus implements CommandBus
{
    private array $handlers = [];

    public function registerHandler(string $commandClass, CommandHandler $handler): void
    {
        $this->handlers[$commandClass] = $handler;
    }

    public function dispatch(Command $command)
    {
        $commandClass = get_class($command);
        if (!isset($this->handlers[$commandClass])) {
            throw new RuntimeException("No handler registered for command: {$commandClass}");
        }

        $handler = $this->handlers[$commandClass];
        return $handler->handle($command);
    }
}

class SimpleQueryBus implements QueryBus
{
    private array $handlers = [];

    public function registerHandler(string $queryClass, QueryHandler $handler): void
    {
        $this->handlers[$queryClass] = $handler;
    }

    public function query(Query $query)
    {
        $queryClass = get_class($query);
        if (!isset($this->handlers[$queryClass])) {
            throw new RuntimeException("No handler registered for query: {$queryClass}");
        }

        $handler = $this->handlers[$queryClass];
        return $handler->handle($query);
    }
}

5. 使用示例

<?php

// 假设你已经配置好了 Doctrine ORM 和 EntityManager

// 初始化 Command Bus 和 Query Bus
$commandBus = new SimpleCommandBus();
$queryBus = new SimpleQueryBus();

// 初始化 Repositories
$writeUserRepository = new DoctrineWriteUserRepository($entityManager);
$readUserRepository = new DoctrineReadUserRepository($entityManager);

// 注册 Command Handlers
$commandBus->registerHandler(CreateUserCommand::class, new CreateUserCommandHandler($writeUserRepository));
$commandBus->registerHandler(UpdateUserEmailCommand::class, new UpdateUserEmailCommandHandler($writeUserRepository));

// 注册 Query Handlers
$queryBus->registerHandler(GetUserByIdQuery::class, new GetUserByIdQueryHandler($readUserRepository));
$queryBus->registerHandler(GetAllUsersQuery::class, new GetAllUsersQueryHandler($readUserRepository));

// 创建用户
$createUserCommand = new CreateUserCommand('John Doe', '[email protected]');
$commandBus->dispatch($createUserCommand);

// 获取用户列表
$getAllUsersQuery = new GetAllUsersQuery();
$users = $queryBus->query($getAllUsersQuery);

// 获取指定用户
$getUserByIdQuery = new GetUserByIdQuery(1);
$user = $queryBus->query($getUserByIdQuery);

数据同步：最终一致性的挑战

CQRS架构中，由于读写分离，可能出现数据不一致的情况。例如，用户创建成功后，读取数据库可能还没有更新。这就是最终一致性问题。

解决最终一致性问题的方法有很多，常见的有：

事件溯源（Event Sourcing）： 将所有状态变更都记录为事件，通过重放事件来重建状态。
消息队列： 使用消息队列异步同步数据，例如Kafka、RabbitMQ等。
补偿事务： 如果写入失败，则执行相应的补偿操作，保证数据一致性。

一些建议和注意事项

不要过度设计： CQRS不是银弹，不要为了用而用。只有在大型应用中，读写压力较大，需要更高的性能和可扩展性时，才考虑使用CQRS。
从小处着手： 可以先在应用的一部分模块中使用CQRS，逐步推广到整个系统。
选择合适的技术栈： 根据业务需求和团队技能，选择合适的技术栈来实现CQRS。例如，可以使用Doctrine ORM、EventSauce、Broadway等框架。
监控和日志： 完善的监控和日志系统可以帮助你及时发现和解决问题。

表格总结

特性	CQRS	传统CRUD
职责	读写分离，Command处理写操作，Query处理读操作	读写操作混合在一起
数据模型	读写数据模型可以不同，针对各自需求进行优化	通常使用相同的数据模型
数据库	可以使用不同的数据库，例如，写数据库使用关系型数据库，读数据库使用NoSQL数据库	通常使用同一个关系型数据库
一致性	最终一致性，需要处理数据同步问题	强一致性
复杂性	较高，引入了更多的组件和概念	较低
适用场景	大型应用，读写压力较大，需要更高的性能和可扩展性	小型应用，读写压力较小，对性能要求不高
优点	性能提升，可扩展性，可维护性，灵活性	简单易懂
缺点	复杂性增加，最终一致性，学习成本	性能瓶颈，扩展困难，代码复杂

最后的话

CQRS是一种强大的架构模式，可以帮助你构建高性能、可扩展、易于维护的大型应用。但是，它也增加了系统的复杂性，需要谨慎使用。希望今天的讲座能让你对CQRS有一个更深入的了解。记住，没有最好的架构，只有最适合的架构。在选择架构模式时，一定要根据实际情况进行权衡，才能做出最佳的选择。

感谢大家的聆听！

发表回复 取消回复

发表回复取消回复