php 使用 rabbitmq 工作模式 work

这里来说 php 使用rabbmitmq 的工作模式 work_queue
work_queue的示意图是这样子的

也就是说有一个生产者，有两个或多个消费者在消费队列，这些消费者之间的关系是竞争的，也就是说一个队列消息，只会有一个消费者使用
这个模式，从图上来看，和简单工作模式差不多，无非就是多了一个消费者，所以，它的代码如下

1.生产者代码

work_queue_sender.php

channel();
    $channel->queue_declare("work_queue",false,false,false,false);
    //因为有两个消费者， 我们这里就使用循环多生产一些消息
    for($i=0;$i<10;$i++){
        $msg = new PhpAmqpLibMessageAMQPMessage($i."----这是生产者生产的第--".$i."--条消息");
        $channel->basic_publish($msg,"","work_queue");
    }
    $channel->close();
    $connection->close();

2.消费者1代码

channel();
    $channel->queue_declare("work_queue",false,false,false,false);
    $callback = function($msg){
        echo "这是消费者1 消费的消息 消息内容是:".$msg->body.PHP_EOL;
    };
    //$channel->basic_consume("work_queue","",false,true,false,false,$callback);
    $channel->basic_consume("work_queue","",false,false,false,false,$callback);
    while(count($channel->callbacks)){
        $channel->wait();
    }

3.消费者2代码

    require __DIR__."/../vendor/autoload.php";

    $connection = new PhpAmqpLibConnectionAMQPStreamConnection('127.0.0.1',5672,"admin","123456");
    $channel = $connection->channel();
    $channel->queue_declare("work_queue",false,false,false,false);
    $callback = function($msg){
        echo "这是消费者1 消费的消息 消息内容是:".$msg->body.PHP_EOL;
    };
    //$channel->basic_consume("work_queue","",false,true,false,false,$callback);
    $channel->basic_consume("work_queue","",false,false,false,false,$callback);
    while(count($channel->callbacks)){
        $channel->wait();
    }

测试，我们先开启两个消费者

再执行生产者

从上面可以看到消费者是交替消费的

ack 机制

我们可以看到

$channel->basic_consume("work_queue","",false,false,false,false,$callback);

上面的代码中 basic_consume 的参数，第四个是false , 它表示开启了消息确认，如果是true, 表示不开启消息确认
为什么有 ack机制，当我们不开启消息确认机制的时候，也就是说，当rabbmitmq 向消费息发送了消息之后，就立即将这条消息从内存中清除，那如果消费者在处理这个消息的时候出现了错误，导至程序停止了，这条消息将会丢失，同时，已经发送给这个消费者的所有消息都将丢失，这是我们不想看到的
那如果我们开启了消息确认，消费者在处理消息后，会向生产者返回消息，这时消息才会被删除，如果程序出现了错误中止，生产者会检测到 tcp连接断开，就会把没有处理过的消息，重新发送给还在运行的另一个消费者进程
这样就不会丢失任何消息了

# 处理消息回调函数  
//首先把第四个参数改成false 
//同时 callback中 添加
//$msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
$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('task_queue', '', false, false, false, false, $callback);

消息的持久化

我们已经学习了如何确保即使消费者异常退出后，任务也不会丢失。但是如果RabbitMQ服务器挂了，我们的任务任然会丢失。

当RabbitMQ退出或崩溃时，它会丢失队列和其中的消息，除非你告诉他要持久化这些信息。需要两件事来确保消息不会丢失：我们需要将队列和消息标记为 durable(持久化)。

首先，我们需要确保RabbitMQ不会丢失消息队列。为了做到这一点，我们需要将队列声明为持久化，只要将 queue_declare 的第三个参数设置为 true 就行了：

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

设置为 true 的标志需要同时应用于生产者和消费者中。
同时，我们还要为我们的消息设置持久化标记

$msg = new AMQPMessage($data,
       array('delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT)
       );

delivery_mode’ => AMQPMessage::DELIVERY_MODE_PERSISTENT 就是把消息也加上了持久化标记

公平调度

RabbitmitMq 不会去看消息是否处理完成，它只会盲目的把消息不断的发送给消费者，这就会出现不平衡的问题，有的消费者计算能力弱，有的计算能力强，此时就可以使用公平调度，等一个消费者处理完消息之后，再给它发送下一条消息
比如在两个消费者的情况下，当所有奇数序号的任务量都很大，并且偶数序号的任务量小的情况下，一个消费者将不断忙碌，另一个消费者几乎没有任何工作量。在这种情境下，RabbitMQ依然不知道什么信息，还是会将消息平均分配。
这种情况的发生是因为RabbitMQ只管分配进入队列的消息。它不会去看消费者的未确认消息的数量。它只是盲目地向第n个消费者发送第n条消息。

为了解决以上问题，我们可以通过设置 basic_qos 第二个参数 prefetch_count = 1。这一项告诉RabbitMQ不要一次给一个消费者发送多个消息。或者换一种说法，在确认前一个消息之前，不要向消费者发送新的消息。相反，新的消息将发送到一个处于空闲的消费者。

$channel->basic_qos(null, 1, null);

最终，综合上面的，我们的生产者代码这样的

channel();
    //这是开启了     消息的持久化
    $channel->queue_declare("work_queue_last",false,true,false,false);
    //因为有两个消费者， 我们这里就使用循环多生产一些消息
    for($i=0;$i<10;$i++){
        $msg = new PhpAmqpLibMessageAMQPMessage($i."----这是生产者生产的第--".$i."--条消息",
        ["delivery_mode"=>PhpAmqpLibMessageAMQPMessage::DELIVERY_MODE_PERSISTENT]);
        $channel->basic_publish($msg,"","work_queue_last");
    }
    $channel->close();
    $connection->close();

消费者的代码是这样的

channel();
    // 因为生产者中的  第三个参数为 true 消费者中对应的也要配成true
    $channel->queue_declare("work_queue_last",false,true,false,false);
    //这里开启 公平调机制
    $channel->basic_qos(null,1,null);
    $callback = function($msg){
        echo "这是消费者1 消费的消息 消息内容是:".$msg->body.PHP_EOL;
        //这里要手机的做 ack
        sleep(3);
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info["delivery_tag"]);
    };
    //这里开启的应答 ack 机制
    $channel->basic_consume("work_queue_last","",false,true,false,false,$callback);
    //$channel->basic_consume("work_queue","",false,false,false,false,$callback);
    while(count($channel->callbacks)){
        $channel->wait();
    }

php 使用 rabbitmq 工作模式 work

Java相关栏目本月热门文章