分类：RabbitMq

消息确认：

消费消息确认模式有：

• AcknowledgeMode.NONE：自动确认
• AcknowledgeMode.AUTO：根据情况确认
• AcknowledgeMode.MANUAL：手动确认
• 1、可以自动创建队列，也可以手动创建队列，如果自动创建队列，那么是谁负责创建队列呢？是生产者？还是消费者？
  
  如果队列不存在，当然消费者不会收到任何的消息。但是如果队列不存在，那么生产者发送的消息就会丢失。所以，为了数据不丢失，消费者和生产者都可以创建队列。那么如果创建一个已经存在的队列呢？那么不会有任何的影响。需要注意的是没有任何的影响，也就是说第二次创建如果参数和第一次不一样，那么该操作虽然成功，但是队列属性并不会改变。队列对于负载均衡的处理是完美的。对于多个消费者来说，RabbitMQ使用轮询的方式均衡的发送给不同的消费者。

  2、RabbitMQ的消息确认机制

  默认情况下，如果消息已经被某个消费者正确的接收到了，那么该消息就会被从队列中移除。当然也可以让同一个消息发送到很多的消费者。

  如果一个队列没有消费者，那么，如果这个队列有数据到达，那么这个数据会被缓存，不会被丢弃。当有消费者时，这个数据会被立即发送到这个消费者，这个数据被消费者正确收到时，这个数据就被从队列中删除。

  那么什么是正确收到呢？通过ack。每个消息都要被acknowledged（确认，ack）。我们可以显示的在程序中去ack，也可以自动的ack。如果有数据没有被ack，那么：

  RabbitMQ Server会把这个信息发送到下一个消费者。

  如果这个app有bug，忘记了ack，那么RabbitMQServer不会再发送数据给它，因为Server认为这个消费者处理能力有限。

  而且ack的机制可以起到限流的作用（Benefitto throttling）：在消费者处理完成数据后发送ack，甚至在额外的延时后发送ack，将有效的均衡消费者的负载。

  消息持久化：

  消息持久化主要是将交换机、队列以及消息设置为durable = true(可持久化的)。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务

RabbitMQ 常用的交换器类型有 fanout direct topic headers 这四种

  1  docker pull rabbitmq:3.7.7-management //management是带管理页面的安装

  2  docker run -d --name rabbitmq3.7.7 -p 5672:5672 -p 15672:15672 -v `pwd`/data:/var/lib/rabbitmq --hostname myRabbit -e RABBITMQ_DEFAULT_VHOST=my_vhost  -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin 2888deb59dfc//最后是image id

测试：

浏览器输入：http://192.168.222.129:15672     默认用户名admin密码admin

php测试rabbitmq

composer安装扩展

"require": {
    "php-amqplib/php-amqplib": ">=2.6.1"
}

新建send.php 端口号为5672

<?php
/**
 * Des 描述
 * Date 2020/8/24
 * Time  10:31
 * User ycl
 */
require 'vendor/autoload.php';
$conn = [
    // Rabbitmq 服务地址
    'host' => '192.168.222.129',
    // Rabbitmq 服务端口
    'port' => '5672',
    // Rabbitmq 帐号
    'login' => 'guest',
    // Rabbitmq 密码
    'password' => 'guest',
    'vhost'=>'/'
];

//创建连接和channel
$conn = new AMQPConnection($conn);
if(!$conn->connect()) {
    die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);

// 用来绑定交换机和队列
$routingKey = 'key_1';

$ex = new AMQPExchange($channel);
//  交换机名称
$exchangeName = 'ex1';
$ex->setName($exchangeName);

// 设置交换机类型
$ex->setType(AMQP_EX_TYPE_DIRECT);
// 设置交换机是否持久化消息
$ex->setFlags(AMQP_DURABLE);
$ex->declare();

for($i=0; $i<5; ++$i){
    echo "Send Message:".$ex->publish(date('H:i:s')."用户".$i."注册" , $routingKey )."\n";
}

新建consume.php 端口号为5672

<?php
/**
 * Des 描述
 * Date 2020/8/24
 * Time  10:32
 * User ycl
 */
require 'vendor/autoload.php';
$conn = [
    // Rabbitmq 服务地址
    'host' => '192.168.222.129',
    // Rabbitmq 服务端口
    'port' => '5672',
    // Rabbitmq 帐号
    'login' => 'guest',
    // Rabbitmq 密码
    'password' => 'guest',
    'vhost'=>'/'
];
try {

//创建连接和channel
    $conn = new AMQPConnection($conn);
    if (!$conn->connect()) {
        die("Cannot connect to the broker!\n");
    }
    $channel = new AMQPChannel($conn);
    $exchangeName = 'ex1';

//创建交换机
    $ex = new AMQPExchange($channel);
    $ex->setName($exchangeName);

    $ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
    $ex->setFlags(AMQP_DURABLE); //持久化
    $ex->declare();

//  创建队列
    $queueName = 'queue1';
    $q = new AMQPQueue($channel);
    $q->setName($queueName);
    $q->setFlags(AMQP_DURABLE);
    $q->declareQueue();

// 用于绑定队列和交换机，跟 send.php 中的一致。
    $routingKey = 'key_1';
    $q->bind($exchangeName, $routingKey);

//接收消息
    $q->consume(function ($envelope, $queue) {
        $msg = $envelope->getBody();
        echo $msg . "\n"; //处理消息
    }, AMQP_AUTOACK);

    $conn->disconnect();
}catch (AMQPException $e){
    echo $e->getMessage();
}

执行

先执行 php consume.php

新开窗口执行 php send.php

结果 consume.php执行窗口会有输出 成功