工作队列

工作队列相比简单队列，就是消费者变成了多个，生产者没有变化，
增加，消息确认机制ACK，confirm机制，消费者在没有确认消息之前，可以处理消息的数量

生产者 send.php

<?php
//引入类库
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
try {
    //创建一个连接
    $connection = new AMQPStreamConnection('127.0.0.1', '5672', 'guest', 'guest', '/');
    //建立通道
    $channel = $connection->channel();
    //confirm推送成功方法
    $channel->set_ack_handler(
        function (AMQPMessage $message) {
            echo "发送成功确认: " . $message->body . PHP_EOL;
        }
    );
    //confirm推送失败方法
    $channel->set_nack_handler(
        function (AMQPMessage $message) {
            echo "发送失败确认: " . $message->body . PHP_EOL;
        }
    );
    //发布确认模式通道变为confirm模式
    $channel->confirm_select();
    //声明需要存储消息的队列，已经存在，可以不用声明
    $channel->queue_declare('work_list', false, false, false, false);
    //指定要发送的消息内容
    for ($i = 0; $i < 10; $i++) {
        $arrar['id'] = $i;
        $arrar['name'] = 'hdiewbewuwewehe';
        $json = json_encode($arrar);
        //消息持久   $msg = new AMQPMessage($json,array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
        $msg = new AMQPMessage($json);
        $channel->basic_publish($msg, '', 'work_list');
//        sleep(1);
    }
    /**
     *您不必在每条消息发送后等待挂起的acks。事实上，这样会更有效率
     *等待尽可能多的邮件被屏蔽。
     * 可以直接执行,也可以放在最后面
     */
    $channel->wait_for_pending_acks();
    // 监听成功或失败返回结束 成功/失败 => set_ack_handler/set_nack_handler
    // 关闭通道
    $channel->close();
    // 关闭连接
    $connection->close();
    echo '发送成功';
} catch (Exception $e) {
    die($e->getMessage());
}

消费者 receive.php

<?php
//引入类库
require_once __DIR__ . '/vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;

//创建一个连接
$connection = new AMQPStreamConnection('127.0.0.1', '5672', 'guest', 'guest', '/'); // 建立连接到RabbitMQ服务器
//建立通道
$channel = $connection->channel();
//声明需要存储消息的队列，已经存在，可以不用声明
//$channel->queue_declare('list_set_1', false, false, false, false);

//消费者执行函数
$callback = function ($msg) {
    echo ' 队列信息： ', $msg->body, "\n";
    sleep(1);
    //开启了自动确认ack，手动发送确认
    $msg->ack();
};
//当前消费者可以同时处理的消息数量
$channel->basic_qos(null, 1, null);
//消费消息->第四个参数 no_ack 是否自动确认ack
$channel->basic_consume('work_list', '', false, false, false, false, $callback);
//当 $channel 有回调时,我们的 $callback 函数都会传递给我们生产者发送的消息,(当队列有消息时,执行回调函数)
while ($channel->is_open()) {
    $channel->wait();
}

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