แพทเทิร์นการเผยแพร่/การสมัครรับข้อมูลของ PHP RabbitMQ (Fanout Exchange)

รูปแบบการเผยแพร่/สับมีขออีกชื่อว่ารูปแบบแฟนเอาท์ใน RabbitMQ สำหรับ PHP

โหมดแฟนเอาท์ใน RabbitMQ เป็นรูปแบบที่ข้อความที่ส่งโดยโปรดิวเซอร์จะถูกประมวลผลโดยผู้บริโภคหลายรายในคิวอันต่าง ๆ ตามที่แสดงในแผนภูมิสถาปัตยกรรมด้านล่าง.

แลกเปลี่ยนแฟนเอาท์สามารถส่งข้อความไปที่คิวที่ผูกกันทั้งหมด.

หมายเหตุ: ไม่ว่าจะใช้โหมดการทำงานของ RabbitMQ แบบใด ความแตกต่างอยู่ที่ประเภทของแลกเปลี่ยนและพารามิเตอร์การเปลี่ยนเส้นทางที่ใช้.

1. บทแนะนำที่ต้องการ

โปรดอ่านบทเรียนต่อไปนี้เพื่อเข้าใจความรู้ที่เกี่ยวข้อง:

• พื้นฐานของ RabbitMQ
• รูปแบบการเผยแพร่/สับมีของ RabbitMQ
• การเริ่มต้นด้วย PHP ใน RabbitMQ (บังคับ โดยที่บทต่อไปจะไม่ทำซ้ำโค้ดแต่จะแสดงโค้ดย่อย)

2. กำหนดแฮชแบบแฟนเอาท์

กำหนดแลกเปลี่ยนผ่านฟังก์ชัน exchange_declare ของช่องข้อมูล.

$channel->exchange_declare(
    'tizi365.fanout', // ชื่อแลกเปลี่ยน ต้องไม่ซ้ำกันและไม่สามารถทำซ้ำได้
    'fanout', // ประเภทแลกเปลี่ยน
    false,
    false, // จะขั้นตอนหรือไม่
    false
);

หมายเหตุ: ทั้งผู้ผลิตข้อความและผู้บริโภคจำเป็นต้องใช้แลกเปลี่ยนนี้.

3. การส่งข้อความ

เราส่งข้อความไปยังแลกเปลี่ยนและแลกเปลี่ยนจะส่งข้อความไปที่คิวที่เกี่ยวข้องตามกฎการเปลี่ยนเส้นทาง.

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

// สร้างการเชื่อมต่อ RabbitMQ
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
// สร้างช่องข้อมูล
$channel = $connection->channel();

// กำหนดแลกเปลี่ยน
$channel->exchange_declare(
    'tizi365.fanout', // ชื่อแลกเปลี่ยน ต้องไม่ซ้ำกันและไม่สามารถทำซ้ำได้
    'fanout', // ประเภทแลกเปลี่ยน
    false,
    false, // จะขั้นตอนหรือไม่
    false
);
// ออบเจกต์ข้อความ โดยใช้เนื้อหาของข้อความเป็นพารามิเตอร์
$msg = new AMQPMessage("สวัสดี tizi365.com");

$channel->basic_publish(
    $msg, // ออบเจกต์ข้อความ
    'tizi365.fanout' // ชื่อแลกเปลี่ยน
);

echo ' [x] ส่งแล้ว ', $msg->getBody(), "\n";

// ปลดทรัพยากร
$channel->close();
$connection->close();

// กำหนดคิว หากชื่อคิวว่างเปล่า ระบบจะสร้าง ID ที่ไม่ซ้ำกันโดยอัตโนมัติและคืนชื่อคิว
list($queue_name, ,) = $channel->queue_declare(
    "", // ชื่อคิว ห้ามซ้ำ หากว่างเปล่า ระบบจะสร้าง ID ที่ไม่ซ้ำกันโดยอัตโนมัติ ทำให้เป็นคิวนิรมาณ
    false,
    false, // บันทึกต่อไปหรือไม่
    true,
    false
);

// ผูกคิวกับแลกเช็นที่กำหนด
$channel->queue_bind(
    $queue_name, // ชื่อคิว
    'tizi365.fanout' // ชื่อแลกเช็น
);

<?php 
require_once __DIR__ . '/vendor/autoload.php';
use PhpAmqpLib\Connection\AMQPStreamConnection;

// สร้างการเชื่อมต่อ rabbitmq
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
// สร้างช่อง
$channel = $connection->channel();

// กำหนดแลกเช็น
$channel->exchange_declare(
    'tizi365.fanout', // ชื่อแลกเช็น ต้องไม่ซ้ำกันและไม่สามารถทำซ้ำได้
    'fanout', // ประเภทของแลกเช็น
    false,
    false, // บันทึกต่อไปหรือไม่
    false
);

// กำหนดคิว
list($queue_name, ,) = $channel->queue_declare(
    "", // ชื่อคิว ห้ามซ้ำ หากว่างเปล่าระบบจะสร้าง ID ที่ไม่ซ้ำกันโดยอัตโนมัติทำให้เป็นคิวนิรมาณ
    false,
    false, // บันทึกต่อไปหรือไม่
    true,
    false
);

// ผูกคิวกับแลกเช็นที่กำหนด
$channel->queue_bind(
    $queue_name, // ชื่อคิว
    'tizi365.fanout' // ชื่อแลกเช็น
);

echo " [*] รอรับข้อความ กด CTRL+C เพื่อออก\n";

// กำหนดฟังก์ชันการประมวลผลข้อความ (ใช้ฟังก์ชันอนุมานที่นี่)
$callback = function ($msg) {
    // ตรรกะการประมวลผลข้อความ
    echo ' [x] ', $msg->body, "\n";
};

// สร้างผู้บริโภค
$channel->basic_consume(
    $queue_name, // ชื่อคิว คิวที่ใช้ในการบริโภค
    '', // แท็กผู้บริโภค ละเว้น ระบบจะสร้าง ID ที่ไม่ซ้ำกัน
    false,
    true, // ต้องการได้รับการยอมรับโดยอัตโนมัติหรือไม่ กล่าวคือแจ้ง rabbitmq ว่าข้อความถูกประมวลผลเรียบร้อย
    false,
    false,
    $callback // ฟังก์ชันการประมวลผลข้อความ
);

// หากช่องยังไม่ปิด ให้บล็อกเพิ่อป้องกันไม่ให้โปรเซสฉกาจออก
while ($channel->is_open()) {
    $channel->wait();
}

// คืนทรัพยากร
$channel->close();
$connection->close();