PHP 微服务与消息队列：RabbitMQ、Kafka 与事件驱动

// 同步（慢：邮件+通知都阻塞请求）
public function checkout(Order $order): void {
    $this->mailer->send($order->customerEmail(), '订单成功');
    $this->notifier->send($order->customerEmail(), '订单更新');
    return ['ok' => true];   // 用户等了 500ms+
}

// 异步（快：只扔队列，立即返回）
public function checkout(Order $order): void {
    OrderEvents::dispatch(new OrderCreated($order));   // 入队
    return ['ok' => true];   // 用户立即收到响应
}

Producer → [Queue] → Consumer
                          ├ 成功 → Ack（删除）
                          ├ 失败 → Nack（重试/死信）
                          └ 崩溃 → 消息未 Ack，broker 重新投递

// 安装：composer require php-amqplib/php-amqplib
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

// 生产者
$conn = new AMQPStreamConnection('rabbitmq', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->queue_declare('orders', durable: true);

$msg = new AMQPMessage(json_encode(['orderId' => 1]), [
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,   // 持久化
]);
$ch->basic_publish($msg, exchange: '', routing_key: 'orders');

// 消费者（常驻进程运行）
$ch->basic_qos(null, 1, null);                     // 每次取一条
$ch->basic_consume('orders', callback: function ($msg) {
    try {
        process(json_decode($msg->getBody(), true));
        $ch->basic_ack($msg->getDeliveryTag());    // 成功才 Ack
    } catch (Throwable $e) {
        $ch->basic_nack($msg->getDeliveryTag(), requeue: false);  // 进死信
    }
});
while ($ch->is_consuming()) { $ch->wait(); }

// 安装：composer require rdkafka（pecl rdkafka）
use RdKafka\Producer;
use RdKafka\KafkaConsumer;
use RdKafka\Message;

// 生产者
$producer = new Producer();
$producer->addBrokers('kafka:9092');
$topic = $producer->newTopic('user-events');
$topic->produce(0, 0, json_encode(['userId' => 1, 'action' => 'signup']));
$producer->flush(1000);

// 消费者（按消费者组 + offset 消费）
$conf = new RdKafka\Conf();
$conf->set('group.id', 'analytics-group');
$conf->set('auto.offset.reset', 'earliest');
$consumer = new KafkaConsumer($conf);
$consumer->subscribe(['user-events']);

while (true) {
    $msg = $consumer->consume(12000);
    if ($msg->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
        handleEvent(json_decode($msg->payload, true));
        $consumer->commit($msg);                 // 提交 offset
    }
}

// Redis Stream 生产
$redis->xAdd('task:email', '*', ['order_id' => 1, 'user_email' => 'a@x.com']);

// 消费（消费者组）
$redis->xGroup('CREATE', 'task:email', 'email-workers', '0', true);
$msgs = $redis->xReadGroup('email-workers', 'worker-1', 'task:email', count: 1, block: 5000);

foreach ($msgs as $msg) {
    process($msg['data']);
    $redis->xAck('task:email', 'email-workers', $msg['id']);   // 确认
}

// 定义 Job（命令模式，见 /php-oop-design-patterns/）
class SendOrderEmail implements ShouldQueue {
    public function __construct(private Order $order) {}
    public function handle(OrderMailer $mailer): void {
        $mailer->send($this->order->customerEmail(), '订单成功');
    }
}

// 分发（同步变异步）
OrderCreated::dispatch($order);                     // 事件
SendOrderEmail::dispatch($order)->onQueue('mail'); // 指定队列

// 重试与退避
class SendOrderEmail implements ShouldQueue {
    public int $tries = 3;
    public array $backoff = [5, 30, 120];   // 依次等待
    public function failed(Order $order, Throwable $e): void {
        // 人工介入：标记、通知
    }
}
// 死信：超过 tries 进 failed_jobs 表

订单服务 ──订单已创建──▶ 通知服务
                 └──▶ 库存服务
                 └──▶ 分析服务

// 订单服务发事件
OrderCreated::dispatch($order);

// 各订阅者各自处理（互不知晓彼此）
// NotificationSubscriber / InventorySubscriber / AnalyticsSubscriber

// 消费者必须幂等（同一消息处理两次结果相同）
function processOrderCreated(array $data): void {
    // 用唯一键做去重（如订单号）
    $key = 'processed:order:' . $data['orderId'];
    if ($redis->get($key)) { return; }
    $redis->setex($key, 86400, 1);
    // ... 业务处理
}
// 或数据库唯一约束兜底