php rabbitmq怎麽開啓confirm模式

Question

$channel->startTransaction();這個是開啓事務的，怎麽開啓confirm模式，求demo?

Answer 1

来个php-amqp的例子：

$message = "It is a test message, publish time:".date("Y-m-d H:i:s");
$ch->confirmSelect();
// 批量发布，一次确认
$messages = [];
for($i=0;$i<10;$i++) {
    $messages[$i] = $i.$message;
    $ex->publish($messages[$i], $routing_key, AMQP_MANDATORY, $params);
}

$ack_callback = function ($delivery_tag, $multiple) {
    // ack处理
    echo 'Message acked', PHP_EOL;
    var_dump(func_get_args());
    return true;
};
$nack_callback = function ($delivery_tag, $multiple, $requeue) use ($messages) {
    // nack处理: 重新发送消息该批次消息，或者记录日志
    echo 'Message nacked', PHP_EOL;
    var_dump(func_get_args());
    return false;
};

$ch->setConfirmCallback($ack_callback, $nack_callback); // 设置回调
$ch->waitForConfirm(1); // 在setConfirmCallback()后调用

Answer 2

给你一个完整的例子

$exchange = 'exchange_name';
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

//异步回调消息确认
$channel->set_ack_handler(
    function (AMQPMessage $message) {

        echo "Message acked with content " . $message->body . PHP_EOL;
    }
);
$channel->set_nack_handler(
    function (AMQPMessage $message) {
        echo "Message nacked with content " . $message->body . PHP_EOL;
    }
);

//设为confirm模式
$channel->confirm_select();

$channel->exchange_declare($exchange, 'fanout', false, false, true);
$msg = 'demo';
$message = new AMQPMessage($msg, array('content_type' => 'text/plain'));
$channel->basic_publish($message, $exchange, null, true);
//阻塞等待消息确认
$channel->wait_for_pending_acks();

$channel->close();
$connection->close();