Swoole 및 RabbitMQ 통합 사례: 고가용성 메시지 대기열 시스템 구축-Swoole-php.cn

인터넷 시대의 도래와 함께 메시지 큐 시스템은 점점 더 중요해졌습니다. 이는 서로 다른 애플리케이션 간의 비동기 작업을 가능하게 하고, 결합을 줄이고, 확장성을 향상시켜 전체 시스템의 성능과 사용자 경험을 향상시킵니다. 메시지 큐잉 시스템에서 RabbitMQ는 다양한 메시지 프로토콜을 지원하며 금융 거래, 전자 상거래, 온라인 게임 및 기타 분야에서 널리 사용되는 강력한 오픈 소스 메시지 큐잉 소프트웨어입니다.

실제 애플리케이션에서는 RabbitMQ를 다른 시스템과 통합해야 하는 경우가 많습니다. 이 기사에서는 swoole 확장을 사용하여 고가용성 RabbitMQ 클러스터를 구현하고 완전한 샘플 코드를 제공하는 방법을 소개합니다.

1. RabbitMQ 통합

RabbitMQ 소개

RabbitMQ는 AMQP(Advanced Message Queuing Protocol) 프로토콜을 완벽하게 따르는 오픈 소스 크로스 플랫폼 메시지 프로토콜입니다. RabbitMQ의 핵심 아이디어는 메시지를 대기열에 넣었다가 필요할 때 꺼내어 효율적인 비동기 데이터 교환 및 통신을 실현하는 것입니다.

RabbitMQ 통합

RabbitMQ를 PHP 애플리케이션과 통합하기 위해 PHP AMQP 라이브러리에서 제공하는 API를 사용할 수 있습니다. 라이브러리는 게시, 구독, 대기열, 교환 및 기타 기능을 포함하여 RabbitMQ의 기본 AMQP 0-9-1 프로토콜 및 확장을 지원합니다. 다음은 간단한 예제 코드입니다.

<?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();
?>

로그인 후 복사

이 예제 코드는 로컬 RabbitMQ 서버('localhost')에 연결하고 'hello'라는 대기열을 선언하고 이 대기열에 메시지를 보냅니다.

2. Swoole 통합

Swoole 소개

Swoole은 EventLoop를 기반으로 비동기 TCP, UDP, HTTP, WebSocket 및 기타 통신 프로토콜을 구현하는 고성능 PHP 비동기 네트워크 통신 프레임워크입니다. 높은 동시성, 고성능, 낮은 소비 및 쉬운 개발이 특징이며 웹 서비스 및 게임 서버와 같은 시나리오에서 널리 사용되었습니다.

Swoole은 RabbitMQ를 통합합니다

Swoole의 비동기 기능은 RabbitMQ 비동기 통신과 매우 호환되며 효율적이고 안정적이며 지연 시간이 짧은 메시지 대기열 시스템을 달성할 수 있습니다. 다음은 RabbitMQ를 통합하는 Swoole의 샘플 코드입니다.

<?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();
?>

로그인 후 복사

이 샘플 코드는 로컬 RabbitMQ 서버('localhost')에 연결하고 영구 대기열 'task_queue'를 선언하고 대기열의 메시지 수신을 시작합니다. 메시지가 도착하면 Swoole은 콜백 함수를 비동기적으로 호출하고 콜백 함수의 비즈니스 로직을 처리한 후 응답을 보내 효율적이고 지연 시간이 짧은 비동기 통신을 달성할 수 있습니다.

3. 고가용성 아키텍처

고가용성 메시지 대기열 시스템을 달성하려면 클러스터에 여러 RabbitMQ 노드를 통합하여 시스템의 확장성과 내결함성을 개선해야 합니다.

일반적으로 사용되는 RabbitMQ 클러스터 구성에는 활성-대기 모드와 미러링 모드가 포함됩니다. 활성-대기 모드에서는 한 노드가 활성 노드 역할을 하고 다른 노드는 백업 노드 역할을 합니다. 기본 노드가 다운되면 백업 노드가 자동으로 그 책임을 맡습니다. 미러 모드에서는 대기열이 여러 노드의 디스크에 복제되고 동기화된 상태로 유지됩니다. 이러한 각 노드는 생산자 및 소비자 요청이 보낸 메시지를 처리할 수 있습니다.

안정성, 확장성, 유지 관리성 및 기타 요소를 고려하여 고가용성 아키텍처로 미러 모드를 선택했습니다. 다음은 구성 파일에 미러 대기열을 추가하기 위한 샘플 코드입니다.

$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'),
));

로그인 후 복사

이 샘플 코드는 'task_queue'라는 영구 대기열을 생성하고 'x-ha-policy' 매개변수를 'all'로 설정하여 이 대기열이 모든 미러 대기열은 "고가용성"입니다. 동시에 'x-dead-letter-exchange' 매개변수도 'dead_exchange'로 설정되는데, 이는 메시지가 거부된 후 이 스위치로 전송된다는 의미입니다. 이 스위치에는 메시지 재사용 또는 통계를 위해 바인딩된 하나 이상의 대기열이 있을 수 있습니다.

4. 전체 샘플 코드

다음은 Swoole 비동기 통신 프레임워크를 사용하여 RabbitMQ 미러 대기열 모드를 통합하여 고가용성 메시지 대기열 시스템을 구현하는 전체 메시지 대기열 시스템 샘플 코드입니다. 실제 필요에 따라 구성이나 코드를 수정하여 고유한 메시지 대기열 시스템을 구현할 수 있습니다.

<?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();
?>

로그인 후 복사

위 코드에서 RabbitMQ에 대한 연결은 먼저 AMQPStreamConnection 클래스를 통해 설정됩니다. 그런 다음 'test.exchange'라는 이름의 스위치와 'test.queue'라는 이름의 대기열을 생성하고 'x-ha-policy'를 'all'로 설정하여 이 대기열이 미러 대기열이고 모든 노드가 액세스할 수 있음을 나타냅니다. 동시에 'x-dead-letter-exchange'도 'dead.exchange'로 설정되는데, 이는 메시지가 거부된 후 'dead.exchange' 스위치로 전송된다는 의미입니다.

마지막으로 콜백 함수에서 basic_ack() 메서드를 사용하여 소비 성공 여부를 확인하고 메시지가 점유한 리소스를 해제합니다.

위는 Swoole과 RabbitMQ의 통합 실습에 대한 관련 내용입니다. Swoole 확장을 사용하면 비동기 통신을 쉽게 구현하고 여러 RabbitMQ 노드를 고가용성 메시지 대기열 시스템에 통합하여 시스템의 성능과 안정성을 향상시킬 수 있습니다.

위 내용은 Swoole 및 RabbitMQ 통합 사례: 고가용성 메시지 대기열 시스템 구축의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!