Swoole and RabbitMQ integration practice: building a high-availability message queue system

With the advent of the Internet era, message queue systems have become more and more important. It enables asynchronous operations between different applications, reduces coupling, and improves scalability, thereby improving the performance and user experience of the entire system. In the message queuing system, RabbitMQ is a powerful open source message queuing software. It supports a variety of message protocols and is widely used in financial transactions, e-commerce, online games and other fields.

In practical applications, it is often necessary to integrate RabbitMQ with other systems. This article will introduce how to use swoole extension to implement a high-availability RabbitMQ cluster and provide a complete sample code.

1. RabbitMQ integration

1. Introduction to RabbitMQ

RabbitMQ is an open source, cross-platform message queue software that fully follows the AMQP protocol (Advanced Message Queuing Protocol) and supports multiple message protocols. The core idea of ​​RabbitMQ is to put messages into the queue and take them out when needed, achieving efficient asynchronous data exchange and communication.

2. RabbitMQ Integration

In order to integrate RabbitMQ with PHP applications, we can use the API provided by the PHP AMQP library. The library supports RabbitMQ's main AMQP 0-9-1 protocol and extensions, including Publish, Subscribe, Queue, Exchange and other functions. Here is a simple sample code:

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

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

// 建立连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 声明队列
$channel->queue_declare('hello', false, false, false, false);

// 创建消息
$msg = new AMQPMessage('Hello World!');

// 发送消息
$channel->basic_publish($msg, '', 'hello');

echo " [x] Sent 'Hello World!'
";

// 关闭连接
$channel->close();
$connection->close();
?>

This sample code connects to the local RabbitMQ server ('localhost'), declares a queue named 'hello' and sends messages to this queue.

2. Swoole integration

1. Swoole introduction

Swoole is a high-performance PHP asynchronous network communication framework that implements asynchronous TCP and UDP based on EventLoop , HTTP, WebSocket and other communication protocols. It is characterized by high concurrency, high performance, low consumption, and easy development. It has been widely used in scenarios such as web services and game servers.

2. Swoole integrates RabbitMQ

Swoole’s asynchronous features are very suitable for RabbitMQ asynchronous communication, and can achieve an efficient, stable, and low-latency message queue system. The following is a sample code for Swoole integrating RabbitMQ:

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

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

// 建立连接
$connection = new AMQPStreamConnection('localhost', 5672, 'guest', 'guest');
$channel = $connection->channel();

// 声明队列
$channel->queue_declare('task_queue', false, true, false, false);

echo " [*] Waiting for messages. To exit press CTRL+C
";

// 接收消息
$callback = function ($msg) {
    echo ' [x] Received ', $msg->body, "
";
    sleep(substr_count($msg->body, '.'));
    echo " [x] Done
";
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);

// 监听消息
while (count($channel->callbacks)) {
    $channel->wait();
}

// 关闭连接
$channel->close();
$connection->close();
?>

This sample code connects to the local RabbitMQ server ('localhost'), declares a persistent queue 'task_queue' and starts listening to the queue's messages. When a message arrives, Swoole will call the callback function asynchronously, and can send a response after processing the business logic in the callback function to achieve efficient, low-latency asynchronous communication.

3. High-availability architecture

In order to achieve a high-availability message queue system, we need to integrate multiple RabbitMQ nodes in a cluster to improve the scalability and fault tolerance of the system.

Commonly used RabbitMQ cluster configurations include active-standby mode and mirroring mode. In the active-standby mode, one node acts as the active node and the other nodes act as backup nodes. When the primary node goes down, the backup node automatically takes over its responsibilities. In mirror mode, a queue is replicated to disk on multiple nodes and kept in sync. Each of these nodes can handle messages sent by producers and consumer requests.

Considering stability, scalability, maintainability and other factors, we chose the mirror mode as our high availability architecture. The following is a sample code for adding a mirror queue in the configuration file:

$channel->queue_declare('task_queue', false, true, false, false, false, array(
    'x-ha-policy' => array('S', 'all'),
    'x-dead-letter-exchange' => array('S', 'dead_exchange'),
));

This sample code creates a persistent queue named 'task_queue' and sets the 'x-ha-policy' parameter to 'all' , indicating that all mirror queues of this queue are "highly available". At the same time, the 'x-dead-letter-exchange' parameter is also set to 'dead_exchange', which means that the message will be sent to this switch after being rejected. This switch can have one or more queues bound for message re-consumption or statistics.

4. Complete sample code

The following is a complete message queue system sample code, which uses the Swoole asynchronous communication framework to integrate the RabbitMQ mirror queue mode to implement a high-availability message queue system. You can modify the configuration or code to implement your own message queue system according to actual needs.

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

use PhpAmqpLibConnectionAMQPStreamConnection;
use PhpAmqpLibMessageAMQPMessage;

$exchangeName = 'test.exchange';
$queueName = 'test.queue';
$deadExchangeName = 'dead.exchange';

// 建立连接
$connection = new AMQPStreamConnection(
    'localhost', 5672, 'guest', 'guest', '/', false, 'AMQPLAIN', null, 'en_US', 3.0, 3.0, null, true
);
$channel = $connection->channel();

// 声明交换机
$channel->exchange_declare($exchangeName, 'direct', false, true, false);

// 声明死信交换机
$channel->exchange_declare($deadExchangeName, 'fanout', false, true, false);

// 声明队列
$channel->queue_declare($queueName, false, true, false, false, false, array(
    'x-ha-policy' => array('S', 'all'),
    'x-dead-letter-exchange' => array('S', $deadExchangeName),
));

// 绑定队列到交换机中
$channel->queue_bind($queueName, $exchangeName);

echo " [*] Waiting for messages. To exit press CTRL+C
";

// 接收消息
$callback = function ($msg) {
    echo ' [x] Received ', $msg->body, "
";
    sleep(substr_count($msg->body, '.'));
    echo " [x] Done
";
    $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};

$channel->basic_qos(null, 1, null);
$channel->basic_consume($queueName, '', false, false, false, false, $callback);

// 监听消息
while (count($channel->callbacks)) {
    $channel->wait();
}

// 关闭连接
$channel->close();
$connection->close();
?>

In the above code, the connection to RabbitMQ is first established through the AMQPStreamConnection class. Then created a switch named 'test.exchange', a queue named 'test.queue', and set 'x-ha-policy' to 'all', indicating that this queue is a mirror queue and all nodes can access. At the same time, ‘x-dead-letter-exchange’ is also set to ‘dead.exchange’, which means that the message will be sent to the ‘dead.exchange’ switch after being rejected.

Finally, in the callback function, use the basic_ack() method to determine the success of consumption and release the resources occupied by the message.

The above is the relevant content about the integration practice of Swoole and RabbitMQ. By using the Swoole extension, we can easily implement asynchronous communication and integrate multiple RabbitMQ nodes into a high-availability message queue system to improve the performance and stability of the system.

The above is the detailed content of Swoole and RabbitMQ integration practice: building a high-availability message queue system. For more information, please follow other related articles on the PHP Chinese website!