PHP Academy · บทเรียน

การทำงานกับ RabbitMQ ใน PHP

เผยแพร่และรับข้อความด้วย RabbitMQ

การทำงานกับ RabbitMQ ใน PHP เป็นบทเรียน PHP Academy ฟรีบน CoddyKit นี่คือบทเรียนที่ 2 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน PHP Academy และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส PHP Academy มีบทเรียนทั้งหมด 4 บทเรียน

RabbitMQ ใน PHP

RabbitMQ คือตัวกลางที่สื่อสารด้วย AMQP 0-9-1 ใน PHP ไคลเอนต์มาตรฐานคือ php-amqplib/php-amqplib (PHP ล้วน) หรือ ext-amqp ที่ทำงานบน C บทเรียนนี้ใช้ php-amqplib เพราะทำงานได้ทุกที่ที่ Composer ทำงาน

โมเดลหลักของ AMQP มีองค์ประกอบสามส่วน: การแลกเปลี่ยนรับข้อความ การผูกกำหนดเส้นทางตามคีย์ และ คิวเก็บข้อความไว้สำหรับผู้บริโภค หากเข้าใจส่วนนี้แล้ว รายละเอียดที่เหลือจะไม่ซับซ้อน

การติดตั้งไคลเอนต์

เพิ่มไลบรารีด้วย Composer ไลบรารีนี้ต้องใช้ส่วนขยาย sockets และ bcmath ซึ่งมักมีอยู่ในการสร้าง PHP CLI สำหรับการใช้งานผ่านบรรทัดคำสั่ง

composer require php-amqplib/php-amqplib
# Connection target, e.g. amqp://guest:guest@localhost:5672/

โมเดลการแลกเปลี่ยน/คิว/การผูก

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

direct — ตรงกับคีย์การกำหนดเส้นทางแบบพอดี
topic — รูปแบบไวลด์การ์ด เช่น order.*.eu
fanout — กระจายไปยังคิวทั้งหมดที่ผูกไว้
headers — จับคู่ตามแอตทริบิวต์ส่วนหัว

การผูกเชื่อมคิวเข้ากับการแลกเปลี่ยนด้วยรูปแบบการกำหนดเส้นทาง การแยกผู้ผลิตออกจากโครงสร้างคิวคือเหตุผลสำคัญของชั้นการแลกเปลี่ยน

การเชื่อมต่อและการประกาศ

เปิดการเชื่อมต่อ รับช่องทาง และประกาศโทโพโลยีของคุณ durable: true ทำให้เอ็กซ์เชนจ์/คิวอยู่รอดหลังโบรกเกอร์เริ่มการทำงานใหม่ การประกาศมีคุณสมบัติทำซ้ำได้อย่างปลอดภัย — การประกาศเอนทิตีที่มีอยู่แล้วด้วยอาร์กิวเมนต์ตรงกันจะไม่ทำอะไร

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

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$ch->exchange_declare('orders', 'topic', false, true, false);
$ch->queue_declare('orders.email', false, true, false, false);
$ch->queue_bind('orders.email', 'orders', 'order.created');

echo "Topology ready\n";
$ch->close();
$conn->close();

การเผยแพร่ข้อความ

ห่อเนื้อหาไว้ใน AMQPMessage ตั้งค่า delivery_mode = 2 เพื่อทำให้ข้อความเป็นข้อความถาวร — คิวแบบคงทนและข้อความแบบถาวรร่วมกันเท่านั้นที่จะอยู่รอดหลังการเริ่มการทำงานใหม่ (มีเพียงอย่างใดอย่างหนึ่งไม่เพียงพอ)

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

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();

$payload = json_encode(['orderId' => 42, 'total' => 19.90]);
$msg = new AMQPMessage($payload, [
    'content_type'  => 'application/json',
    'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
    'message_id'    => bin2hex(random_bytes(8)),
]);

$ch->basic_publish($msg, 'orders', 'order.created');
echo "Published\n";
$ch->close();
$conn->close();

การรับข้อความด้วยการยืนยันด้วยตนเอง

ลงทะเบียนฟังก์ชันเรียกกลับด้วย basic_consume ส่งค่า no_ack = false เพื่อให้คุณควบคุมการยืนยันได้ ให้ยืนยันเฉพาะหลังงานสำเร็จเท่านั้น หากล้มเหลว ให้ใช้ basic_nack พร้อมเปิดการส่งกลับเข้าคิว เพื่อให้ RabbitMQ ส่งข้อความซ้ำ

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

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->basic_qos(null, 10, null); // prefetch 10

$callback = function ($msg) {
    $data = json_decode($msg->getBody(), true);
    try {
        // ... do work ...
        $msg->ack();
    } catch (\Throwable $e) {
        $msg->nack(true); // requeue
    }
};

$ch->basic_consume('orders.email', '', false, false, false, false, $callback);
while ($ch->is_consuming()) {
    $ch->wait();
}

การดึงข้อความล่วงหน้าและการกระจายอย่างเป็นธรรม

ตามค่าเริ่มต้น RabbitMQ จะส่งข้อความให้ผู้รับข้อความแบบวนรอบโดยไม่คำนึงว่าผู้รับแต่ละรายกำลังทำงานหนักเพียงใด — ผู้รับข้อความที่ทำงานช้าจะมีงานสะสม basic_qos(null, prefetch, null) จำกัดจำนวนข้อความที่ยังไม่ได้รับการยืนยันซึ่งผู้รับข้อความรายหนึ่งถือไว้ได้

กำหนดค่า prefetch เป็นจำนวนน้อย (เช่น 1–10) สำหรับงานหนักที่ใช้เวลาไม่เท่ากัน เพื่อให้โบรกเกอร์ส่งงานใหม่เฉพาะให้ผู้รับข้อความที่มีความจุว่าง วิธีนี้เรียกว่าการกระจายอย่างเป็นธรรม

การยืนยันจากผู้เผยแพร่

basic_publish ส่งคืนทันทีและไม่ได้บอกว่าโบรกเกอร์รับข้อความแล้วหรือไม่ สำหรับการเผยแพร่ที่รับประกันได้ ให้เปิดใช้การยืนยันจากผู้เผยแพร่: โบรกเกอร์จะส่งการยืนยันเมื่อข้อความถูกจัดเก็บหรือกำหนดเส้นทางอย่างปลอดภัยแล้ว

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

$conn = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->confirm_select(); // enable confirms on this channel

$ch->set_ack_handler(fn($m) => print("confirmed\n"));
$ch->set_nack_handler(fn($m) => print("REJECTED\n"));

$ch->basic_publish(new AMQPMessage('hi'), 'orders', 'order.created');
$ch->wait_for_pending_acks(5.0); // block until confirmed or timeout

การจัดการข้อความตาย

กำหนดอาร์กิวเมนต์ x-dead-letter-exchange ของคิว เพื่อให้ข้อความที่ถูกปฏิเสธ (nack โดยไม่ส่งกลับเข้าคิว) หรือข้อความที่หมดอายุถูกส่งไปยัง DLX ใช้ร่วมกับ x-delivery-limit บนคิวโควรัมเพื่อจำกัดจำนวนการลองใหม่โดยอัตโนมัติ

<?php
use PhpAmqpLib\Wire\AMQPTable;

$args = new AMQPTable([
    'x-dead-letter-exchange'    => 'orders.dlx',
    'x-dead-letter-routing-key' => 'order.failed',
    'x-message-ttl'             => 60000, // ms before expiry
]);
// false=passive, true=durable, false=exclusive, false=autodelete, args
$ch->queue_declare('orders.email', false, true, false, false, false, $args);
$ch->queue_declare('orders.dead', false, true, false, false);
$ch->queue_bind('orders.dead', 'orders.dlx', 'order.failed');

การเรียกใช้ผู้ปฏิบัติงานในระบบจริง

กฎด้านการปฏิบัติงานที่ผ่านการใช้งานจริงมาอย่างยากลำบากสำหรับผู้ปฏิบัติงาน RabbitMQ ใน PHP:

เรียกใช้ผู้รับข้อความเป็นกระบวนการ CLI ที่ทำงานต่อเนื่องเป็นเวลานานภายใต้ตัวควบคุม (systemd / Supervisor) ซึ่งจะเริ่มกระบวนการใหม่เมื่อกระบวนการสิ้นสุด
PHP ใช้หน่วยความจำเพิ่มขึ้นเรื่อย ๆ เมื่อเวลาผ่านไป — เริ่มผู้ปฏิบัติงานใหม่หลังรับข้อความครบ N ข้อความ หรือเมื่อหน่วยความจำถึงเกณฑ์ที่กำหนด
ส่งสัญญาณชีพของ AMQP และจัดการ SIGTERM เพื่อปิดการทำงานอย่างสุภาพ (ทำข้อความที่กำลังดำเนินการอยู่ให้เสร็จ จากนั้นจึงหยุด)
ใช้คิวโควรัมสำหรับ HA แทนคิวแบบจำลองเก่า

การปิดการทำงานอย่างสุภาพ

การนำระบบขึ้นใช้งานจะส่ง SIGTERM ผู้ปฏิบัติงานที่ไม่ได้เตรียมการจะหยุดกลางข้อความ ทำให้ต้องส่งข้อความนั้นซ้ำ ติดตั้งตัวจัดการสัญญาณที่เปลี่ยนสถานะ แล้วทำข้อความปัจจุบันให้เสร็จ ยืนยันข้อความ จากนั้นจึงออกจากลูปรับข้อความอย่างเรียบร้อย pcntl_async_signals(true) ช่วยให้ PHP ส่งสัญญาณระหว่างการรอของ AMQP ได้

<?php
pcntl_async_signals(true);
$running = true;
pcntl_signal(SIGTERM, function () use (&$running) {
    $running = false; // stop after the current message
    echo "SIGTERM: draining...\n";
});

while ($running && $ch->is_consuming()) {
    try {
        $ch->wait(null, false, 5); // wakes for signals
    } catch (\PhpAmqpLib\Exception\AMQPTimeoutException $e) {
        // idle tick - loop and re-check $running
    }
}
$ch->close();
echo "Stopped cleanly\n";

ตรวจสอบอย่างรวดเร็ว

การอยู่รอดหลังโบรกเกอร์เริ่มการทำงานใหม่

สรุปทบทวน

ขณะนี้คุณสามารถสร้างกระบวนการทำงาน RabbitMQ ใน PHP ที่ใช้งานได้จริง:

เผยแพร่ไปยังเอ็กซ์เชนจ์ แล้วกำหนดเส้นทางผ่านการผูกไปยังคิว
คิวแบบคงทนและข้อความแบบถาวรจะอยู่รอดหลังการเริ่มการทำงานใหม่
รับข้อความด้วยการยืนยันด้วยตนเอง และปรับ prefetch ของ basic_qos เพื่อการกระจายอย่างเป็นธรรม
ใช้การยืนยันจากผู้เผยแพร่สำหรับการส่งที่รับประกันได้ และใช้DLXสำหรับข้อความที่ไม่สามารถประมวลผลได้
เรียกใช้ผู้ปฏิบัติงานภายใต้ตัวควบคุม พร้อมสัญญาณชีพและการปิดการทำงานอย่างสุภาพ

ถัดไป: Kafka เมื่อคุณต้องการการสตรีมที่มีปริมาณงานสูงแทนการจัดคิวงาน

คำถามที่พบบ่อย

บทเรียน “การทำงานกับ RabbitMQ ใน PHP” ฟรีหรือไม่

ใช่ — ข้อความเต็มของ “การทำงานกับ RabbitMQ ใน PHP” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส PHP Academy ให้อัปเกรดเป็น CoddyKit PRO คอร์ส PHP Academy มีบทเรียนทั้งหมด 4 บทเรียน

คุณจะเรียนรู้อะไรในบทเรียน “การทำงานกับ RabbitMQ ใน PHP”

เผยแพร่และรับข้อความด้วย RabbitMQ คุณปฏิบัติ PHP Academy ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน

คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน PHP Academy หรือไม่

ไม่จำเป็นต้องมีประสบการณ์มาก่อน PHP Academy บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 2 จากทั้งหมด 4 บทเรียน

บทเรียน “การทำงานกับ RabbitMQ ใน PHP” ใช้เวลานานแค่ไหน

บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย

ฉันเขียนและรันโค้ดในบทเรียน PHP Academy นี้ได้ไหม

ได้ บทเรียน PHP Academy ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ

บทเรียนทั้งหมดในหลักสูตรนี้

เหตุผลที่ควรใช้การส่งข้อความแบบอะซิงโครนัส
การทำงานกับ RabbitMQ ใน PHP
Apache Kafka กับ PHP
การสร้างเวิร์กโฟลว์ที่ขับเคลื่อนด้วยเหตุการณ์

← กลับไปที่ PHP Academy