Praca z RabbitMQ w PHP
Proszę publikować i odbierać komunikaty za pomocą RabbitMQ.
Praca z RabbitMQ w PHP to bezpłatna lekcja PHP Academy na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej PHP Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs PHP Academy zawiera 4 lekcji w sumie.
RabbitMQ w PHP
RabbitMQ to broker obsługujący protokół AMQP 0-9-1. W PHP standardowym klientem jest php-amqplib/php-amqplib (czysty PHP) albo wspierane przez C rozszerzenie ext-amqp. W tej lekcji używana jest biblioteka php-amqplib, ponieważ działa wszędzie tam, gdzie działa Composer.
Podstawowy model AMQP obejmuje trzech uczestników: exchanges odbierają wiadomości, bindings kierują je według klucza, a queues przechowują je dla konsumentów. Po opanowaniu tego modelu reszta to szczegóły.
Instalowanie klienta
Bibliotekę należy dodać za pomocą Composera. Wymaga ona rozszerzeń sockets i bcmath, które są powszechnie dostępne w kompilacjach PHP przeznaczonych do pracy w CLI.
composer require php-amqplib/php-amqplib
# Connection target, e.g. amqp://guest:guest@localhost:5672/Model exchange/queue/binding
Producenci publikują wiadomości w exchange, nigdy bezpośrednio w kolejce. Typ exchange decyduje o routingu:
direct— dokładne dopasowanie klucza routingu.topic— wzorce z symbolami wieloznacznymi, takie jakorder.*.eu.fanout— rozgłaszanie do wszystkich powiązanych kolejek.headers— dopasowanie na podstawie atrybutów nagłówków.
Binding łączy kolejkę z exchange za pomocą wzorca routingu. Odsprzężenie producentów od topologii kolejek jest właśnie celem warstwy exchange.
Nawiązywanie połączenia i deklarowanie
Należy nawiązać połączenie, uzyskać kanał i zadeklarować topologię. durable: true sprawia, że exchange i kolejka przetrwają restart brokera. Deklaracje są idempotentne — ponowne zadeklarowanie istniejącej encji z identycznymi argumentami niczego nie zmienia.
<?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();Publikowanie wiadomości
Należy owinąć treść w obiekt AMQPMessage. Ustawienie delivery_mode = 2 sprawia, że wiadomość jest trwała — to połączenie trwałej kolejki i trwałej wiadomości pozwala przetrwać restart (jedno bez drugiego nie wystarczy).
<?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();Odbieranie z ręcznym potwierdzaniem
Należy zarejestrować callback za pomocą basic_consume. Należy przekazać no_ack = false, aby samodzielnie sterować potwierdzaniem. Wiadomość należy potwierdzić dopiero po pomyślnym zakończeniu pracy; w razie błędu użycie basic_nack z opcją requeue pozwala RabbitMQ dostarczyć ją ponownie.
<?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();
}Prefetch i sprawiedliwe rozdzielanie
Domyślnie RabbitMQ rozdziela wiadomości między konsumentów metodą round-robin, nie uwzględniając obciążenia każdego z nich — powolny konsument gromadzi pracę. basic_qos(null, prefetch, null) ogranicza liczbę niepotwierdzonych wiadomości, które może przechowywać konsument.
W przypadku ciężkich i nierównomiernych zadań należy ustawić prefetch na niewielką liczbę (np. 1–10), aby broker wysyłał nowe zadania tylko konsumentom, którzy mają wolne zasoby. To właśnie sprawiedliwe rozdzielanie.
Potwierdzenia publikowania
basic_publish zwraca wynik natychmiast i nie informuje, czy broker zaakceptował wiadomość. Aby zapewnić niezawodne publikowanie, należy włączyć potwierdzenia publikowania: broker wysyła ack, gdy wiadomość zostanie bezpiecznie zapisana lub skierowana do odbiorcy.
<?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 timeoutPrzekierowywanie wiadomości odrzuconych
Należy skonfigurować argument x-dead-letter-exchange kolejki, aby odrzucone wiadomości (nack bez requeue) lub wiadomości, których ważność wygasła, były kierowane do DLX. W przypadku kolejek kworum można połączyć to z x-delivery-limit, aby automatycznie ograniczyć liczbę ponowień.
<?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');Uruchamianie workerów w produkcji
Kilka sprawdzonych zasad operacyjnych dotyczących workerów PHP RabbitMQ:
- Należy uruchamiać konsumentów jako długotrwałe procesy CLI pod kontrolą supervisora (systemd / Supervisor), który ponownie uruchomi je po zakończeniu pracy.
- PHP z czasem zużywa coraz więcej pamięci — należy restartować workera po przetworzeniu N wiadomości albo po przekroczeniu określonego progu pamięci.
- Należy wysyłać AMQP heartbeaty i obsługiwać
SIGTERM, aby zapewnić kontrolowane zamykanie (dokończyć przetwarzanie bieżącej wiadomości, a następnie się zatrzymać). - W celu zapewnienia wysokiej dostępności należy używać kolejek kworum zamiast starszych kolejek lustrzanych.
Kontrolowane zamykanie
Podczas wdrażania wysyłany jest sygnał SIGTERM. Naiwny worker kończy działanie w połowie przetwarzania wiadomości, co wymusza jej ponowne dostarczenie. Należy zainstalować handler sygnałów, który ustawia flagę; następnie dokończyć bieżącą wiadomość, potwierdzić ją i poprawnie przerwać pętlę odbierania. pcntl_async_signals(true) pozwala PHP dostarczać sygnały między oczekiwaniami na 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";Szybkie sprawdzenie
Przetrwanie restartu brokera.
Podsumowanie
Można już zbudować w PHP kompletny potok RabbitMQ:
- Publikowanie do exchange i kierowanie wiadomości przez powiązania do kolejek.
- Trwała kolejka + trwała wiadomość przetrwają restart.
- Odbieranie z ręcznym potwierdzaniem i dostrajanie wartości prefetch w basic_qos w celu zapewnienia sprawiedliwego rozdzielania.
- Używanie potwierdzeń publikowania dla niezawodnego wysyłania oraz DLX dla wiadomości, których nie można poprawnie przetworzyć.
- Uruchamianie workerów pod kontrolą supervisora, z heartbeatami i kontrolowanym zamykaniem.
Następny temat: Kafka, gdy potrzebna jest strumieniowa transmisja o dużej przepustowości zamiast kolejkowania zadań.
Często zadawane pytania
Czy lekcja „Praca z RabbitMQ w PHP” jest bezpłatna?
Tak — pełny tekst „Praca z RabbitMQ w PHP” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu PHP Academy, przejdź na CoddyKit PRO. Kurs PHP Academy zawiera 4 lekcji w sumie.
Co nauczysz się w „Praca z RabbitMQ w PHP”?
Proszę publikować i odbierać komunikaty za pomocą RabbitMQ. Ćwiczysz PHP Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć PHP Academy?
Nie wymagamy żadnego doświadczenia. PHP Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.
Ile czasu zajmuje lekcja „Praca z RabbitMQ w PHP”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji PHP Academy?
Tak. Każda lekcja PHP Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Dlaczego warto używać komunikacji asynchronicznej
- Praca z RabbitMQ w PHP
- Apache Kafka z PHP
- Tworzenie przepływów pracy sterowanych zdarzeniami