Webman uses RabbitMQ messaging middleware to implement system asynchronous decoupling practical tutorial

Introduction

RabbitMQ is an open source message broker software that uses the Advanced Message Queuing Protocol (AMQP) to send and receive messages. RabbitMQ supports multiple message protocols, including STOMP, MQTT, etc., and can be integrated with multiple programming languages and platforms, such as Java, .NET, Python, etc.

AMQP, or Advanced Message Queuing Protocol, is a network protocol and an open standard for application layer protocols designed for message-oriented middleware. Clients and message middleware based on this protocol can transmit messages without being restricted by different client/middleware products, different development languages, and other conditions.

Basic architecture design

Image source: https://blog.csdn.net/cuierdan/article/details/123824300

Basic Concepts

Message Broker: (message broker server) is a virtual concept, and RabbitMQ is an instance of Message Broker.
Producer: A program that generates data and sends it to a message broker is called a message producer.
Connection: (Connection) The connection created by the producer and consumer with the message broker server (Message Broker) through the TCP protocol.
Channel: A virtual connection created in a Connection, similar to the concept of a connection pool when connecting to a database. Producers and consumers do not communicate directly with MQ through a Connection, but through a Channel, and data flows in the Channel.
VirtualHost: (Virtual Message Server) is like the concept of database instance in MySQL database, and can specify the user's permission to operate the library and table. It can also be classified into different users in LINUX system, and different users are independent of each other. Each VirtualHost is equivalent to a relatively independent mini RabbitMQ server. Each VirtualHost is isolated from each other, and exchange, queue, message, etc. cannot communicate with each other.
Exchange: (Switch) The switch is directly connected to the Channel, receives data from the message producer, and then routes the message to one or more Queues (or discards it). The Exchange does not store messages. RabbitMQ switches are of four types: fanout, direct, topic, and headers. Each switch type corresponds to different routing rules. According to different routing rules, the switch will route messages to different queues.
Binding: (Binding) A virtual connection between an exchange and a queue. In this binding, you can set a Binding Key. A binding is to connect an exchange and a queue with a Binding Key. There are certain rules for setting the Binding Key. Exchange will match the Routing Key carried in the message with the rules set in the Binding Key and send the message to the corresponding queue. The Binding information is saved in the query table in the Exchange and is used as the basis for Exchange to distribute messages to queues.
Routing Key: (Routing Key) is used to match the basis for routing rules. When a producer sends a message to an Exchange, it generally specifies a Routing Key. The Exchange will match the routing rules set in the Binding according to the Routing Key and send messages that meet the rules to the specified queue.
Queue: (Message queue) An internal object in RabbitMQ that is used as a container for storing messages. RabbitMQ will send messages in the queue to consumers in one of the six modes of RabbitMQ. RabbitMQ will send messages in the queue to one or more consumers depending on the selected mode. Before connecting to the consumer, the message is waiting for the consumer to take the message from the queue.
Consumer: A message consumer, representing an application that takes messages from a queue.

Features

Reliability: RabbitMQ uses some mechanisms to ensure reliability, such as persistence, transmission confirmation, and publication confirmation.
Flexible routing: Routes messages through switches before they enter queues.
Scalability: Multiple RabbitMQ nodes can form a cluster, and the nodes in the cluster can also be dynamically expanded according to actual business conditions.
High availability: Queues can be mirrored across machines in a cluster, so that the queue remains available even if some nodes fail.
Multiple protocols: In addition to natively supporting the AMQP protocol, RabbitMQ also supports multiple message middleware protocols such as STOMP and MQTT.
Support multi-language clients: RabbitMQ supports almost all common languages, such as Java, Python, Ruby, PHP, C#, JavaScript, etc.

Key Features

Message queue: Allows an application to send messages to a queue, which are then taken from the queue and processed by another application.
Message Routing: Supports routing messages from a sender to one or more receivers.
Message persistence: Ensure that messages are not lost after a system failure.
Message confirmation: Ensure that the message is processed correctly and can be resent if processing fails.
Cluster: Supports running on multiple nodes to provide high availability and load balancing.

Install

1Panel is used for installation here. 1Panel is a modern, open source Linux server operation and maintenance management panel.

App Store

picture

Install the image

picture

Note: You need to check [Port External Access] to facilitate local debugging

Mirror

After the installation is successful, you can find the installed RabbitMQ image container in the image

picture

External network accessible address: http://{{your public network ip}}:15672. If it is on a cloud server, remember to open port 15672 in the security policy.

picture

RabbitMQ management interface port: 15672. It is a web application used to manage and monitor RabbitMQ message brokers
AMQP default port: 5672. It is a network protocol used to transmit messages between applications, usually used in message queue systems.

Console

picture

use

Here we use the webman plug-in RabbitMQ client, plug-in address: https://www.workerman.net/plugin/67,

Thank you very much for the plugin contribution from Mr. Rabbit! 🌻

Plugin Features

Supports 5 consumption modes: simple queue, workQueue, routing, pub/sub, exchange;
Support delay queue (RabbitMQ plug-in must be installed);
Asynchronous non-blocking consumption, asynchronous non-blocking production, synchronous blocking production

Install the plugin

Install via composer package management:

 composer require workbunny/webman-rabbitmq

Note: Before installing this plug-in, please make sure you have installed the webman framework. Related installation documents: https://www.workerman.net/doc/webman/install.html

Configuration

All plugin configuration files path: config/plugin/workbunny/webman-rabbitmq/app.php

 <?php return [ 'enable' => true, 'host' => '120.120.120.74', 'vhost' => '/', 'port' => 5672, 'username' => 'rabbitmq', 'password' => 'rabbitmq', 'mechanisms' => 'AMQPLAIN', ... ];

Change host to the server public IP
port changed to 15672

consumer

Here, the command is used to create a RestyBuilder with a single-process consumer

 ./webman workbunny:rabbitmq-builder resty --mode=queue ℹ️ Config updated. ℹ️ Builder created. ✅ Builder RestyBuilder created successfully.

After the creation is complete, the complete consumer file location is process/workbunny/rabbitmq/RestyBuilder.php

 <?php declare(strict_types=1); namespace process\workbunny\rabbitmq; use Bunny\Channel as BunnyChannel; use Bunny\Async\Client as BunnyClient; use Bunny\Message as BunnyMessage; use Workbunny\WebmanRabbitMQ\Constants; use Workbunny\WebmanRabbitMQ\Builders\QueueBuilder; class RestyBuilder extends QueueBuilder { /** * @var array = [ * 'name' => 'example', * 'delayed' => false, * 'prefetch_count' => 1, * 'prefetch_size' => 0, * 'is_global' => false, * 'routing_key' => '', * ] */ protected array $queueConfig = [ // 队列名称，默认由类名自动生成'name' => 'process.workbunny.rabbitmq.RestyBuilder', // 是否延迟'delayed' => false, // QOS 数量'prefetch_count' => 0, // QOS size 'prefetch_size' => 0, // QOS 全局'is_global' => false, // 路由键'routing_key' => '', ]; /** @var string 交换机类型*/ protected string $exchangeType = Constants::DIRECT; /** @var string|null 交换机名称,默认由类名自动生成*/ protected ?string $exchangeName = 'process.workbunny.rabbitmq.RestyBuilder'; /** @inheritDoc */ public function handler(BunnyMessage $message, BunnyChannel $channel, BunnyClient $client): string { echo '[RabbitMQ][队列消费] Tag：' .$message->consumerTag. PHP_EOL; echo '[RabbitMQ][队列消费着] Content：' .$message->content. PHP_EOL; echo '[RabbitMQ][队列消费着] Exchange：' .$message->exchange. PHP_EOL; return Constants::ACK; } }

Start webman

 php start.php start Workerman[start.php] start in DEBUG mode ----------------------------------------------------------------------- WORKERMAN ----------------------------------------------------------------------- Workerman version:4.1.15 PHP version:8.2.18 Event-Loop:\Workerman\Events\Event ------------------------------------------------------------------------ WORKERS ------------------------------------------------------------------------ proto user worker listen processes status tcp root webman http://0.0.0.0:8217 2 [OK] tcp root monitor none 1 [OK] tcp root plugin.workbunny.webman-rabbitmq.process.workbunny.rabbitmq.RestyBuilder none 1 [OK] ---------------------------------------------------------------------------------------------------------------------------------------------------------

Producer

 use function Workbunny\WebmanRabbitMQ\sync_publish; use process\workbunny\rabbitmq\RestyBuilder; sync_publish(RestyBuilder::instance(), '兔子大佬你好呀！'); # return bool

Consumption results

 [RabbitMQ][队列消费] Tag：process.workbunny.rabbitmq.RestyBuilder [RabbitMQ][队列消费着] Content：开源技术小栈你好呀！ [RabbitMQ][队列消费着] Exchange：process.workbunny.rabbitmq.RestyBuilder ... [RabbitMQ][队列消费] Tag：process.workbunny.rabbitmq.RestyBuilder [RabbitMQ][队列消费着] Content：兔子大佬你好呀！ [RabbitMQ][队列消费着] Exchange：process.workbunny.rabbitmq.RestyBuilder

picture