消息確認機制

前言

消息隊列的下游，業務邏輯可能復雜，處理任務可能花費很長時間。若在一條消息到達它的下游，任務剛處理了一半，由於不確定因素，下游的任務處理進程

被kill掉啦，導致任務無法執行完成。而沿用我們前面幾章的消息刪除【消息一旦拋給下游，就立馬從隊列刪除】，這可能會引發問題——消息沒有處理完，但是隊列

里的消息已經被刪除了。

因此，rabbitmq內含消息確認機制【Message acknowledgment】，簡稱ack。rabbitmq將消息發送給consumer，此刻消息不會從隊列刪除，consumer消費完畢消息后，

給rabbitmq發送一條ack，在rabbitmq收到ack后，才從隊列里刪除。

代碼

在consumer層的代碼，需要修改兩步。

一、添加ack代碼

$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);

二、開啟消息確認機制

$channel->basic_consume('task_queue', '', false, false, false, false, $callback); //basic_consume第四個參數置成false，no_ack=false

整體代碼見下

生產者sender.php

<?php
/**
 * sender.php
 * Created by PhpStorm.
 * User: wangdaxi
 * Date: 2017/10/18
 * Time: 14:26
 */
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
$channel = $connection->channel();


$channel->queue_declare('hello', false, false, false, false);

$data = implode(" ", array_slice($argv, 1));
empty($data) && $data = "Hello World!";

$msg = new AMQPMessage($data);

$channel->basic_publish($msg, '', 'hello');

echo " [x] Sent '$data'\n";

//close the channel and connection;
$channel->close();
$connection->close();

消費者receive.php

<?php
/**
 * receive.php
 * Created by PhpStorm.
 * User: wangdaxi
 * Date: 2017/10/18
 * Time: 14:34
 */
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'guest', 'guest');
$channel = $connection->channel();

$channel->queue_declare('hello', false, false, false, false);
echo ' [*] Waiting for messages. To exit press CTRL+C', "\n";

$callback = function($msg) {
    echo "[x] Received ", $msg->body, "\n";
    sleep(substr_count($msg->body, '.'));
    echo "[x] Done\n";
    //消息確認
    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};
$channel->basic_consume('hello', '', false, false, false, false, $callback);

while(count($channel->callbacks)) {
    $channel->wait();
}