So implementieren Sie verteilte Nachrichtenwarteschlangen und Broadcasts in PHP-Microservices

So implementieren Sie verteilte Nachrichtenwarteschlangen und Broadcasts in PHP-Mikrodiensten

Vorwort:
In der modernen verteilten Systementwicklung sind Nachrichtenwarteschlangen und Broadcasts sehr häufige Komponenten, mit denen die Kommunikation zwischen verschiedenen Systemen, Entkopplung und Kommunikation implementiert wird. In der PHP-Microservice-Architektur können wir zur Implementierung verteilter Nachrichtenverarbeitungs- und Broadcast-Funktionen einige ausgereifte Open-Source-Tools und Frameworks verwenden, um die Entwicklung zu vereinfachen. In diesem Artikel wird die Verwendung von RabbitMQ und Swoole zur Implementierung verteilter Nachrichtenwarteschlangen und Broadcasts vorgestellt.

1. Grundlegende Konzepte und Verwendung von RabbitMQ
RabbitMQ ist eine zuverlässige, plattformübergreifende Open-Source-Messaging-Middleware. Es folgt dem AMQP-Standard (Advanced Message Queuing Protocol) und bietet umfassende Funktionen zur Nachrichtenproduktion und -nutzung. Im Folgenden sind einige grundlegende Konzepte von RabbitMQ aufgeführt:

1. Produzent: das Programm, das Nachrichten sendet.
2. Warteschlange: Ein Container, der Nachrichten speichert.
3. Consumer: Ein Programm, das Nachrichten empfängt und verarbeitet.
4. Verbraucherbestätigungen: Nachdem der Verbraucher die Nachricht erhalten hat, sendet er eine Bestätigungsnachricht an die Warteschlange, um die Warteschlange darüber zu informieren, dass die Nachricht verarbeitet wurde.
5. Exchange: Empfängt von Produzenten gesendete Nachrichten und leitet die Nachrichten gemäß bestimmten Regeln an die Warteschlange weiter.
6. Binding: Bindung der Beziehung zwischen dem Austauscher und der Warteschlange.

Das Folgende ist ein Beispiel-PHP-Code, der zeigt, wie man Nachrichten in RabbitMQ sendet und empfängt:

// 创建连接
$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 "Sent 'Hello World!'";

// 接收消息
$callback = function ($msg) {
  echo "Received: ", $msg->body, "
";
};

$channel->basic_consume('hello', '', false, true, false, false, $callback);

while ($channel->is_consuming()) {
  $channel->wait();
}

// 关闭通道和连接
$channel->close();
$connection->close();

2. Die grundlegenden Konzepte und Verwendung von Swoole
Swoole ist ein leistungsstarkes Netzwerkkommunikations-Framework auf Basis von PHP, das leistungsstarkes Asynchronous bietet E/A-Funktionen und ereignisgesteuertes Programmiermodell. In der PHP-Microservice-Architektur können wir Swoole verwenden, um verteilte Nachrichten-Broadcast-Funktionen zu implementieren.

Im Folgenden sind einige Grundkonzepte von Swoole aufgeführt:

1. Server: Ein Programm, das Netzwerkanfragen empfängt und verarbeitet.
2. Client: Das Programm, das Netzwerkanfragen sendet.
3. Ereignis: Die Interaktion zwischen dem Server und dem Client.
4. Asynchron: Eine Methode, die die Ausführung des Hauptprozesses nicht blockiert.
5. Synchron: Eine Methode zum Blockieren der Ausführung des Hauptprozesses, bis der Vorgang abgeschlossen ist.

Das Folgende ist ein Beispiel-PHP-Code, der zeigt, wie man einen TCP-Server erstellt und Nachrichten in Swoole sendet:

// 创建服务器
$server = new swoole_server("127.0.0.1", 9501);

// 注册事件回调函数
$server->on('connect', function ($server, $fd) {
  echo "Client {$fd}: connect.
";
});

$server->on('receive', function ($server, $fd, $from_id, $data) {
  echo "Received: $data 
";

  // 广播消息给所有客户端
  $server->sendtoAll($data);
});

$server->on('close', function ($server, $fd) {
  echo "Client {$fd}: close.
";
});

// 启动服务器
$server->start();

3. Implementieren Sie verteilte Nachrichtenwarteschlangen in PHP-Microservices.
Um verteilte Nachrichten in PHP-Microservices Queue zu implementieren, können wir Verwenden Sie RabbitMQ und Swoole zusammen. Zuerst müssen wir einen RabbitMQ-Consumer und einen Swoole-TCP-Server starten.

RabbitMQ-Verbrauchercodebeispiel:

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

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

// 设置每次只接收一条消息
$channel->basic_qos(null, 1, null);

// 定义消息处理的回调函数
$callback = function ($msg) {
  echo "Received: ", $msg->body, "
";
  // 模拟任务处理
  sleep(3);
  echo "Task finished.
";
  // 显示确认消息
  $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
};

// 监听队列，接收消息
$channel->basic_consume('task_queue', '', false, false, false, false, $callback);

while ($channel->is_consuming()) {
  $channel->wait();
}

// 关闭通道和连接
$channel->close();
$connection->close();

Swoole TCP-Servercodebeispiel:

// 创建服务器
$server = new swoole_server("127.0.0.1", 9501);

$server->set([
  'worker_num' => 4,         // 设置工作进程数
  'task_worker_num' => 4,    // 设置任务进程数
]);

// 注册事件回调函数
$server->on('connect', function ($server, $fd) {
  echo "Client {$fd}: connect.
";
});

$server->on('receive', function ($server, $fd, $from_id, $data) {
  echo "Received: $data 
";
  
  // 将接收到的消息发送给任务进程处理
  $server->task($data);
});

$server->on('task', function ($server, $task_id, $from_id, $data) {
  // 模拟任务处理
  sleep(3);
  
  // 处理结果发送给请求进程
  $server->finish($data);
});

$server->on('finish', function ($server, $task_id, $data) {
  // 将处理结果发送给客户端
  $server->send($data);
});

$server->on('close', function ($server, $fd) {
  echo "Client {$fd}: close.
";
});

// 启动服务器
$server->start();

Wenn der RabbitMQ-Verbraucher die Nachricht empfängt, bedeutet dies, dass eine Aufgabe erstellt wird und mit der Verarbeitung beginnt. Anschließend sendet der Swoole TCP-Server die empfangene Nachricht zur Verarbeitung an den Aufgabenprozess und sendet das Verarbeitungsergebnis über die Rückruffunktion an den Client.

4. Implementieren Sie die verteilte Nachrichtenübertragung in PHP-Mikrodiensten.
Um die verteilte Nachrichtenübertragung in PHP-Mikrodiensten zu implementieren, können wir die Broadcast-Funktion von Swoole mit dem verteilten Cache (z. B. Redis) kombinieren. Zuerst müssen wir einen Swoole TCP-Server und einen Redis-Abonnenten erstellen.

Codebeispiel für Swoole TCP-Server:

// 创建服务器
$server = new swoole_server("127.0.0.1", 9501);

// 注册事件回调函数
$server->on('connect', function ($server, $fd) {
  echo "Client {$fd}: connect.
";
});

$server->on('receive', function ($server, $fd, $from_id, $data) {
  echo "Received: $data 
";

  // 将接收到的消息广播给所有客户端
  $server->sendtoAll($data);
});

$server->on('close', function ($server, $fd) {
  echo "Client {$fd}: close.
";
});

// 启动服务器
$server->start();

Codebeispiel für Redis-Abonnenten:

// 创建Redis连接
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// 订阅消息
$redis->subscribe('channel', function ($redis, $channel, $message) {
  echo "Received from Redis: $message 
";
  
  // 发送消息给Swoole TCP服务器
  $client = new swoole_client(SWOOLE_SOCK_TCP);
  if (!$client->connect('127.0.0.1', 9501, -1)) {
    echo "Failed to connect to server.";
    exit;
  }
  $client->send($message);
  $client->close();
});

Wenn Redis die Nachricht empfängt, sendet es sie über die Rückruffunktion an den Swoole TCP-Server, und der Server sendet die empfangene Nachricht dann an alle Kunden.

Zusammenfassung:
Durch den obigen Beispielcode können wir lernen, wie man RabbitMQ und Swoole verwendet, um verteilte Nachrichtenwarteschlangen- und Broadcast-Funktionen in PHP-Mikrodiensten zu implementieren. Diese Technologien und Tools können uns dabei helfen, leistungsstarke und skalierbare verteilte Systeme aufzubauen und die Systementkopplung und -zuverlässigkeit zu verbessern.

Das obige ist der detaillierte Inhalt vonSo implementieren Sie verteilte Nachrichtenwarteschlangen und Broadcasts in PHP-Microservices. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!