場景

之前開發(fā)一個(gè)電競比分網(wǎng)系統(tǒng)，有許多模塊依賴實(shí)時(shí)比賽狀態(tài)（待開始、進(jìn)行中、已結(jié)束、異常），比賽狀態(tài) 進(jìn)行中->已結(jié)束 由圖像識別處理，識別到比賽結(jié)束后向消息隊(duì)列發(fā)送某場比賽的狀態(tài)信息，其他模塊只需訂閱隊(duì)列消息獲取比賽狀態(tài)更新并進(jìn)行對于邏輯處理

RabbitMQ 概念

• 交換器(Exchanges)

  RabbitMQ消息傳遞模型的核心思想是，生產(chǎn)者不發(fā)送任何信息直接到隊(duì)列。事實(shí)上，生產(chǎn)者甚至不知道消息是否會發(fā)送到任何隊(duì)列。生產(chǎn)者只能向交換器發(fā)送消息（也叫交換機(jī)，默認(rèn)交換器使用""空字符標(biāo)記）。交換器需要知道如何處理接收的消息，將消息推入到指定的隊(duì)列中，決定消息是否入列和拋棄。如下圖，P表示消息發(fā)布者，X表示交換機(jī)，Q1和Q2表示不同的隊(duì)列

  
  
  rabbitmq-exchanges

• 交換類型

• direct：
  消息中的路由鍵（routing key）如果和 Binding 中的 binding key 一致， 交換器就將消息發(fā)到對應(yīng)的隊(duì)列中。路由鍵與隊(duì)列名完全匹配，如果一個(gè)隊(duì)列綁定到交換機(jī)要求路由鍵為"dog"，則只轉(zhuǎn)發(fā) routing key 標(biāo)記為"A1"的消息，不會轉(zhuǎn)發(fā)"A2"，也不會轉(zhuǎn)發(fā)"A3"等等。它是完全匹配、單播的模式

• fanout：
  廣播訂閱，向所有的消費(fèi)者發(fā)布消息。每個(gè)發(fā)到 fanout 類型交換器的消息都會分到所有綁定的隊(duì)列上去。fanout 交換器不處理路由鍵，只是簡單的將隊(duì)列綁定到交換器上，每個(gè)發(fā)送到交換器的消息都會被轉(zhuǎn)發(fā)到與該交換器綁定的所有隊(duì)列上。很像子網(wǎng)廣播，每臺子網(wǎng)內(nèi)的主機(jī)都獲得了一份復(fù)制的消息(fanout 類型轉(zhuǎn)發(fā)消息是最快的)

• topic：
  交換器通過模式匹配分配消息的路由鍵屬性，將路由鍵和某個(gè)模式進(jìn)行匹配，此時(shí)隊(duì)列需要綁定到一個(gè)模式上。它將路由鍵和綁定鍵的字符串切分成單詞，這些單詞之間用點(diǎn)隔開。

• 兩個(gè)通配符：符號"#"和符號"*"

  • #：匹配0個(gè)或多個(gè)單詞
  • *：匹配不多不少一個(gè)單詞

RabbitMQ 安裝運(yùn)行

• 使用Docker安裝RabbitMQ

$ docker pull rabbitmq:3.8.3-management

• 運(yùn)行
• 服務(wù)端口：5672
• 管理端端口：15672

$ docker run --name rabbitmq -d -p 5672:5672 -p 15672:15672 -v /data:/var/lib/rabbitmq rabbitmq:3.8.3-management

• web管理端登錄查看(http://127.0.0.1:15672)

默認(rèn)賬號：guest，默認(rèn)密碼：guest

rabbitmq-admin

• 添加管理員
• 命令行

$ docker exec -it 89e8e968aebc bash
root@89e8e968aebc:/# rabbitmqctl add_user ar414 ar414 
root@89e8e968aebc:/# rabbitmqctl set_user_tags ar414 administrator 

• Web管理端

  
  
  rabbitmq-add-admin

• 添加vhost

  
  
  rabbitmq-add-vhost

PHP 簡單使用

安裝

$ composer require php-amqplib/php-amqplib

發(fā)布者

<?php

require_once __DIR__ . '/../vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;

$exchange = 'Gaming';

$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'ar414', 'ar414', 'test');
$channel = $connection->channel();
$channel->exchange_declare($exchange, 'direct', false, false, false);
for ($i = 0; $i < 100; $i++) {
    $routes = ['dota', 'csgo', 'lol'];
    $key = array_rand($routes);
    $arr = [
        'match_id' => $i,
        'status' => rand(0,3)
    ];
    $data = json_encode($arr);
    $msg = new AMQPMessage($data);

    $channel->basic_publish($msg, $exchange, $routes[$key]);
    echo '發(fā)送 '.$routes[$key].' 消息: ' . $data . PHP_EOL;
}
$channel->close();
$connection->close();

訂閱者

<?php

require_once __DIR__ . '/../vendor/autoload.php';

use PhpAmqpLib\Connection\AMQPStreamConnection;
$exchange = 'Gaming';
$routerKey = 'lol'; //只訂閱LOL消息

$connection = new AMQPStreamConnection('127.0.0.1', 5672, 'ar414', 'ar414', 'test');
$channel = $connection->channel(); $channel->exchange_declare($exchange, 'direct', false, false, false);
list($queueName, ,) = $channel->queue_declare("", false, false, true, false);
$channel->queue_bind($queueName, $exchange, $routerKey);

echo " 等待消息中..." .PHP_EOL;
$callback = function ($msg) {
    echo '接收到消息：',$msg->delivery_info['routing_key'], ':', $msg->body, PHP_EOL;
    sleep(1);  //模擬耗時(shí)執(zhí)行
};
$channel->basic_consume($queueName, '', false, true, false, false, $callback);

while ($channel->is_consuming()) {
    $channel->wait();
}

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

運(yùn)行

1. 運(yùn)行某一個(gè)訂閱者程序監(jiān)聽LOL消息隊(duì)列(LolSub.php)
2. 運(yùn)行發(fā)送者程序(Send.php)

發(fā)送者

$ php Send.php 

rabbitmq-send

ar414

LOL訂閱者

$ php LolSub.php 

lol-sub