利用PHP訊息佇列開發高並發訂單處理系統的實作方法

<?php
require_once 'vendor/autoload.php';

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

// 创建连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 声明消息队列
$channel->queue_declare('order_queue', false, true, false, false);

// 创建生产者对象
$producer = new PhpAmqpLibChannelAMQPChannel($channel);

// 发布消息
$message = new AMQPMessage('order content');
$producer->basic_publish($message, '', 'order_queue');

// 关闭连接
$channel->close();
$connection->close();

<?php
require_once 'vendor/autoload.php';

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

// 创建回调函数
function process_order(AMQPMessage $message)
{
    // 处理订单消息
    $order = $message->body;

    // TODO: 订单处理相关操作

    // 手动确认消息已处理
    $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
}

// 创建连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 声明消息队列
$channel->queue_declare('order_queue', false, true, false, false);

// 创建消费者对象
$consumer = new PhpAmqpLibChannelAMQPChannel($channel);

// 设置回调函数
$consumer->basic_consume('order_queue', '', false, false, false, false, 'process_order');

// 持续监听消息队列
while (count($channel->callbacks)) {
    $channel->wait();
}

// 关闭连接
$channel->close();
$connection->close();