RabbitMQ installeren, configureren en gebruiken

Message brokers zijn tussenliggende applicaties die worden gebruikt in verschillende software-architecturen, zoals microservices. Ze verzenden informatie in de vorm van berichten van de ene service naar de andere.

Een veelgebruikt concept dat verband houdt met message brokers is het publisher-subscriber-model, waarbij gebeurtenissen worden verzonden door de zogenaamde publishers (Producers) naar consumers (Consumers).

Meestal neemt een tussenliggende component — een message broker — deel aan de implementatie van dit model.

Om het eenvoudig te maken, gebruiken we een analogie. Denk aan YouTube: er zijn kanalen waarop je je kunt abonneren om meldingen te ontvangen. Wanneer er een nieuwe video wordt gepubliceerd (een gebeurtenis), ontvang je een melding omdat je bent geabonneerd.

Hoe applicaties met message brokers werken

Drie hoofdtermen zijn hier belangrijk: Exchange, Queue en Binding.

• Exchange. Dit is waar berichtgebeurtenissen naartoe worden gestuurd. De Exchange beslist welke gebeurtenissen naar welke wachtrij gaan, op basis van verbindingen — bindings — die we zo meteen zullen bespreken.

• Queue. De wachtrij is een gegevensstructuur die werkt volgens het principe "first in, first out" (FIFO). Ze slaat gegevens op schijf of in het geheugen op en vertegenwoordigt direct de berichten. De wachtrij levert kopieën van deze berichten aan de consumers.

• Binding. Dit verwijst naar de reeks regels die de Exchange vertellen naar welke wachtrijen berichten moeten worden gerouteerd. Er kunnen meerdere bindings bestaan met verschillende parameters tussen een Exchange en een Queue.

Andere belangrijke termen:

• Producer is een service die berichten genereert en deze naar de broker verzendt.
• Consumer is een andere service die begint met verwerken zodra een bericht wordt ontvangen. In onze YouTube-analogie is de publisher YouTube zelf, en de consumer is je telefoon die pushmeldingen ontvangt.

Hier volgt een stapsgewijze uitleg van hoe het proces werkt:

1. De Producer stuurt een bericht dat naar de Exchange gaat.

2. De Queue slaat dit bericht op, bijvoorbeeld op schijf.

3. Wanneer een Consumer klaar is om een bericht te ontvangen (omdat het vorige is verwerkt of de eventverwerkingsapp is gestart), stuurt de server een kopie van de gegevens uit de wachtrij naar de consumer.

4. De Consumer ontvangt het bericht, verwerkt het en stuurt vervolgens een bevestiging (ACK) terug naar de broker.

5. Nadat de broker de bevestiging heeft ontvangen, wordt het bericht verwijderd.

Wat is RabbitMQ

RabbitMQ is een populaire open-source message broker die is gebouwd op het AMQP (Advanced Message Queuing Protocol), een open protocol voor het verzenden van gebeurtenisberichten via een speciale broker. AMQP biedt een breed scala aan mogelijkheden en wordt, naast RabbitMQ, ook geïmplementeerd door andere systemen zoals Apache Qpid.

RabbitMQ is geschreven in de programmeertaal Erlang, en de belangrijkste voordelen zijn de hoge verwerkingssnelheid en maximale routeringsflexibiliteit. RabbitMQ biedt talloze configuratie-opties om regels te definiëren die bepalen welke berichten waarheen worden verzonden voor verwerking.

Verbindingen met RabbitMQ worden gemaakt via het TCP-protocol: de client start een verbinding met het hostadres en houdt deze open zolang interactie met RabbitMQ nodig is. RabbitMQ ondersteunt ook authenticatie, zoals gebruikersnaam en wachtwoord, voor veilige toegang.

Installatie op verschillende besturingssystemen

RabbitMQ bestaat uit de server en de WebUI (beheerinterface). Met de WebUI kun je controleren wat er binnen de broker gebeurt, zoals of de nodes in het cluster actief zijn, hoeveel berichten in verwerking zijn, en meer. Laten we bekijken hoe je RabbitMQ installeert op verschillende besturingssystemen.

Voordat we beginnen, moeten we de hostnaam van de server bijwerken. Voer het volgende commando uit:

sudo hostnamectl set-hostname <nieuwe-hostnaam>

Bijvoorbeeld, we gebruiken rabbitmq als hostnaam voor deze server:

sudo hostnamectl set-hostname rabbitmq

Installeer nano:

sudo apt install nano

En voeg de nieuwe hostnaam toe aan het bestand /etc/hosts:

sudo nano /etc/hosts

Het zou er als volgt uit moeten zien:

127.0.0.1 <nieuwe-hostnaam>

In ons voorbeeld:

127.0.0.1 rabbitmq

Ubuntu / Debian

We installeren RabbitMQ op Ubuntu 22.04. Voor andere Ubuntu/Debian-distributies, raadpleeg de handleiding op de officiële website.

1. Werk de pakketlijst bij en installeer de afhankelijkheden:

sudo apt-get update -y
sudo apt-get install curl gnupg -y
sudo apt-get install apt-transport-https

2. Voeg de ondertekeningssleutels van de repository toe:

curl -1sLf "https://keys.openpgp.org/vks/v1/by-fingerprint/0A9AF2115F4687BD29803A206B73A36E6026DFCA" | sudo gpg --dearmor | sudo tee /usr/share/keyrings/com.rabbitmq.team.gpg > /dev/null
curl -1sLf https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-erlang.E495BB49CC4BBE5B.key | sudo gpg --dearmor | sudo tee /usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg > /dev/null
curl -1sLf https://github.com/rabbitmq/signing-keys/releases/download/3.0/cloudsmith.rabbitmq-server.9F4587F226208342.key | sudo gpg --dearmor | sudo tee /usr/share/keyrings/rabbitmq.9F4587F226208342.gpg > /dev/null

3. Maak het repositorybestand (Apt Source List):

sudo tee /etc/apt/sources.list.d/rabbitmq.list <<EOF
deb [arch=amd64 signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa1.rabbitmq.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa1.rabbitmq.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
deb [arch=amd64 signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa2.rabbitmq.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.E495BB49CC4BBE5B.gpg] https://ppa2.rabbitmq.com/rabbitmq/rabbitmq-erlang/deb/ubuntu jammy main
deb [arch=amd64 signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa1.rabbitmq.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa1.rabbitmq.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
deb [arch=amd64 signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa2.rabbitmq.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
deb-src [signed-by=/usr/share/keyrings/rabbitmq.9F4587F226208342.gpg] https://ppa2.rabbitmq.com/rabbitmq/rabbitmq-server/deb/ubuntu jammy main
EOF

4. Werk opnieuw de pakketlijst bij:

sudo apt-get update -y

5. Installeer Erlang:

sudo apt-get install -y erlang-base \
                       erlang-asn1 erlang-crypto erlang-eldap erlang-ftp erlang-inets \
                       erlang-mnesia erlang-os-mon erlang-parsetools erlang-public-key \
                       erlang-runtime-tools erlang-snmp erlang-ssl \
                       erlang-syntax-tools erlang-tftp erlang-tools erlang-xmerl

6. Installeer vervolgens RabbitMQ:

sudo apt-get install rabbitmq-server -y --fix-missing

Controleer na de installatie of RabbitMQ-server actief is:

systemctl status rabbitmq-server

Maak een aangepaste gebruiker aan:

sudo rabbitmqctl add_user <gebruikersnaam> <wachtwoord>

Wijs de benodigde rechten toe:

sudo rabbitmqctl set_user_tags  <gebruikersnaam> administrator
sudo rabbitmqctl set_permissions -p /  <gebruikersnaam> ".*" ".*" ".*"

Activeer nu de RabbitMQ-beheerconsole:

sudo rabbitmq-plugins enable rabbitmq_management

Ga naar [je server-IP]:15672. Voer je gebruikersnaam en wachtwoord in om toegang te krijgen tot de RabbitMQ WebUI.

Windows

De installatiehandleiding is beschikbaar op de website van RabbitMQ. Je kunt Chocolatey gebruiken of de standaardinstaller.

Voordat je RabbitMQ installeert, moet je eerst Erlang/OTP installeren vanaf de officiële Erlang-website.

Voer het gedownloade bestand uit en volg de installatieprocedure.

Na het installeren van Erlang kun je RabbitMQ installeren met de officiële installer.

Activeer daarna de WebUI:

cd 'C:\Program Files\RabbitMQ Server\rabbitmq_server-3.13.0\sbin'
./rabbitmq-plugins.bat enable rabbitmq_management

Je kunt nu naar localhost:15672 gaan met de standaardgebruikersnaam en -wachtwoord guest:guest.

Docker

Installeren via Docker is waarschijnlijk de eenvoudigste en meest handige manier om RabbitMQ te installeren.

Je kunt RabbitMQ installeren met de volgende opdracht:

docker run --rm -p 15672:15672 rabbitmq:3.13.7-management

De beheerinterface is beschikbaar op poort 15672, waar je kunt inloggen met de standaardgegevens guest:guest.

Deze methode is echter niet geschikt voor productieomgevingen. Docker kent een servernaam toe aan de container, en de broker slaat zijn status op in een map met die naam. Bij elke nieuwe containerbuild zal RabbitMQ de statusinformatie verliezen.

We kunnen alle benodigde serviceparameters eenvoudig definiëren met docker-compose, zoals het wijzigen van de standaardaanmeldingsgegevens of het koppelen van een map voor statuspersistentie. Hier is een voorbeeld van een docker-compose.yml-bestand:

version: "3.3"
services:
  rabbit:
    image: rabbitmq:3.13.7-management
    environment:
      - RABBITMQ_DEFAULT_USER=admin #enter your username
      - RABBITMQ_DEFAULT_PASS=password #enter your password
    volumes:
      - ./rabbit:/var/lib/rabbitmq # mounting folder for state persistence
    ports:
      - 15672:15672 # expose port for the admin interface

Voer uit:

docker compose up -d

De RabbitMQ WebUI is nu beschikbaar.

Overzicht van RabbitMQ Management

Op de hoofdpagina van de beheerinterface van RabbitMQ kun je algemene informatie bekijken, zoals nodes, hun status, het totale aantal berichten en meer.

• Connections – de lijst met verbindingen met het cluster.

• Channels – de lijst met kanalen. Binnen één verbinding kunnen meerdere kanalen worden aangemaakt waarlangs berichten worden verzonden, om te voorkomen dat meerdere TCP-verbindingen worden gemaakt.

Exchanges

Laten we de Exchanges van dichterbij bekijken.

Er bestaan verschillende soorten Exchanges, elk met een eigen filtermechanisme voor berichten. Op basis van deze regels worden gebeurtenissen doorgestuurd naar specifieke wachtrijen.

Direct Exchange

Elk bericht heeft een sleutel die Routing Key wordt genoemd. Berichten worden doorgestuurd naar wachtrijen waarvan de binding dezelfde sleutel opgeeft.

Topic Exchange

De routering gebeurt op basis van een patroon. Bij het maken van het patroon kun je nul of meer woorden (letters en cijfers) gebruiken die door punten worden gescheiden, bijvoorbeeld key.event, en de symbolen # en *.

Fanout Exchange

Er worden geen filterregels toegepast. Elk bericht dat naar een Fanout Exchange wordt gestuurd, wordt naar alle wachtrijen doorgestuurd.

Headers Exchange

Gebruikt berichtheaders en vergelijkt sleutel-waardeparen met de bindingheaders.

Queues

Wanneer je een specifieke wachtrij bekijkt, zie je een grafiek met het aantal berichten in de wachtrij en statistieken over levertijden en verwerkingstijden.

Berichten kunnen twee statussen hebben:

• Ready – wachtend op verwerking.
• Unacked – wordt verwerkt door de consumer.

Publisher-Subscriber implementeren in Python

Laten we bekijken hoe je RabbitMQ gebruikt om het Pub-Sub-model te implementeren in Python. Je moet Python geïnstalleerd hebben (in dit voorbeeld versie 3.11.5) en de Pika-bibliotheek.

1. Maak of selecteer een map waarin de applicatiecode zal staan:

mkdir rabbitmq-article

2. Open deze map in je IDE (bijv. Visual Studio Code of PyCharm).

3. Installeer de Pika-bibliotheek:

pip install pika

4. Maak twee bestanden aan: sender.py en receiver.py.

sender.py-code

from pika import BlockingConnection, ConnectionParameters
from pika.exchange_type import ExchangeType

# Create connection, specifying parameters in the ConnectionParameters object
connection = BlockingConnection(ConnectionParameters(host='localhost'))

# Create a channel
channel = connection.channel()

# Declare an exchange
channel.exchange_declare('new_exchange', ExchangeType.direct)

# Declare a queue
queue = channel.queue_declare(queue='new_queue')

# Bind the queue to the exchange
channel.queue_bind(exchange='new_exchange', queue='new_queue', routing_key='key')

# Publish a message
channel.basic_publish(exchange='new_exchange', routing_key='key', body='Hello World!')

print("Message 'Hello World!' sent")
connection.close()

In de bovenstaande sender.py-code:

• We maken een verbinding met RabbitMQ op localhost.

• We declareren een directe exchange genaamd new_exchange.

• We declareren een wachtrij genaamd new_queue.

• We koppelen de wachtrij aan de exchange met behulp van de routeringssleutel key.

• Ten slotte publiceren we het bericht 'Hello World!' naar de exchange met de opgegeven routeringssleutel.

receiver.py-code

from pika import BlockingConnection, ConnectionParameters
from pika.exchange_type import ExchangeType
import sys, os

def main():
    # Create connection
    connection = BlockingConnection(ConnectionParameters(host='localhost'))
    # Create channel
    channel = connection.channel()

    # Declare an exchange
    channel.exchange_declare('new_exchange', ExchangeType.direct)

    # Declare a queue
    queue = channel.queue_declare(queue='new_queue')

    # Bind the queue to the exchange
    channel.queue_bind(exchange='new_exchange', queue='new_queue', routing_key='key')

    # Function to handle incoming messages
    def handle(ch, method, properties, body):
        print(f"Received message: {body.decode()}")

    # Bind the callback function and queue
    channel.basic_consume(queue='new_queue', on_message_callback=handle, auto_ack=True)

    print('Waiting for messages. Press Ctrl+C to exit.')
    channel.start_consuming()

if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

In de bovenstaande receiver.py-code:

• We maken een verbinding en kanaal met RabbitMQ, net als in de sender.

• We declareren de directe exchange (new_exchange) en de wachtrij (new_queue).

• De wachtrij wordt gekoppeld aan de exchange met de routeringssleutel key.

• De functie handle() verwerkt de berichten door ze af te drukken zodra ze worden ontvangen.

• De methode basic_consume() abonneert de callback-functie op de wachtrij.

• We gebruiken start_consuming() om het proces te laten wachten op berichten voor onbepaalde tijd.

Code uitvoeren

Start eerst de receiver:

python receiver.py

Je zou moeten zien:

Waiting for messages. Press Ctrl+C to exit.

Start vervolgens de sender:

python sender.py

Uitvoer:

Message 'Hello World!' sent

In de terminal waar de receiver draait, zie je:

Received message: Hello World!

In de RabbitMQ-beheerconsole kun je zien dat de exchange new_exchange is gekoppeld aan de wachtrij new_queue met de routeringssleutel key.

Conclusie

In dit artikel hebben we onderzocht wat message brokers zijn, hoe applicaties ermee communiceren en hoe je RabbitMQ installeert en gebruikt als message broker. We hebben met succes een publisher-subscriber-model geïmplementeerd met Python, RabbitMQ en de Pika-bibliotheek, waarbij de zender berichten publiceert naar een exchange en de ontvanger ze verwerkt.