.NET Microservices mit RabbitMQ

Frank Pommerening

• Senior - Softwareentwickler
• Consultant
• Softwarearchitekt

 @FPommerening

RabbitMQ - ein Überblick

Was ist RabbitMQ?

• Message Broker / Queue Manager
  • System um Queues zu verwalten
  • Nachrichten von Systemen empfangen
  • Nachrichten zwischenspeichern
  • Nachrichten nach Regeln an Systeme zustellen
• Erstellt in Erlang / OTP
• Erlaubt Entkopplung von unterschiedlichen und unabhängigen Anwendungen

Nachrichtentransport

12 Begriffe sollt ihr sein

AMQP (Advanced Message Queuing Protocol)
Protokoll zur Kommunikation über Systemgrenzen

Produzent (Producer)
Anwendung, die eine Nachricht erstellt

Konsument (Consumer)
Anwendung, die eine Nachricht empfängt

Nachricht (Message)
Information, die zwischen Produzent und Konsument ausgetauscht wird

Warteschlange (Queue)
Speichert die Nachrichten zwischen

Verbindung (Connection)
TCP Netzwerkverbindung zwischen Anwendung und RabbitMQ broker

Channel
Virtuelle Verbindung innerhalb der TCP-Verbindung

Binding
Verknüpfung zwischen Queue und Exchange

Routing key
"Adresse", nach welcher der Exchange entscheidet, in welche Queue er die Nachricht leitet.

Benutzer (User)

• Zugangsdaten (Benutzername / Password) für RabbitMQ
• Zuordnung von Rechten (Lesen, Schreiben, Konfigurieren etc.)
• Definition global oder für einen spezifischen virtuellen Host

Vhost (virtual host)
Erlaubt, Anwendung auf einer RabbitMQ Instanz zu isolieren

Exchange

• Fanout Exchange
  • Eingehende Nachrichten werden an alle verbunden Queues gesendet
  • Keine Beachtung der Binding-Konfiguration
• Direct Exchange
  • Producer erstellt Routing Key
  • Binding Key der Queue muss exakt passen
• Topic
  • Ähnlich dem Direct Exchange
  • Einsatz von Platzhaltern, z.B. Wort (*) / mehrere Worte (#)
• Header Exchange
  • Routing anhand von Headerinformationen der Nachricht

RabbitMQ - Installation

Installation Windows

Installation Linux

                            
                                apt-get install rabbitmq-server 
                            
                        

Management Plugin

Docker

Starten RabbitMQ Instanz

                            
                                $ docker run -d --hostname my-rabbit --name firstRabbit rabbitmq:3
                            
                        

                            
                                $ docker run -d --hostname myRabbit --name myRabbit 
-p 5672:5672 -p 15672:15672 rabbitmq:3-management
                            
                        

EasyNetQ

• Einfach
• Open Source
• Modular

Warum nicht direkt

var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare(exchange: "logs", type: "fanout");
    var queueName = channel.QueueDeclare().QueueName;
    channel.QueueBind(queue: queueName, exchange: "logs", 
        routingKey: "");
    var consumer = new EventingBasicConsumer(channel);
    consumer.Received += (model, ea) =>{
        var message = Encoding.UTF8.GetString(ea.Body);
        DoSomething(MyMessage.Deserialize(message));    };
    channel.BasicConsume(queue: queueName, noAck: true,
     consumer: consumer);
}

var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare(exchange: "logs", type: "fanout");
    var message = new MyMessage {Content = "Hallo Welt"};
    var body = Encoding.UTF8.GetBytes(message.Serialize());
    channel.BasicPublish(exchange: "logs", routingKey: "",
            basicProperties: null, body: body);
}

Vorteile

• Messaging Pattern, z.B. Publish/Subscribe
• Routing Strategien
• Serialisierung / Deserialisierung als JSON (lesbar)
• Handling von Thread für Konsumenten
• Subscriber Reconnect
• QoS / publisher confirm
• Fehlerbehandlung

API Design

EasyNetQ Einstieg

Verbindungsaufbau

using System;
using EasyNetQ;

namepace FP.MsRmq.Connecting
{
    public class Programm
    {
        public static void Main(string[] args)
        {
            var myBus = RabbitHutch.CreateBus("host=myRabbitMQ");
        }
    }
}

                    

Verbindungsoptionen

Publish / Subscribe

 myBus.Subscribe<MyClass>("MySub", 
        msg => DoSomething(msg));
 myBus.SubscribeAsync<MyClass>("MySubAsync",
        msg => DoSomethingAsync(msg));
     
                    

var msg = new MyClass{};                            
myBus.Publish(msg);
myBus.PublishAsync(msg);
     
                    

Topic Based Routing

 myBus.Subscribe<MyClass>("MyTopicSub", 
        msg => DoSomething(msg), x => x.WithTopic("MyTopic"));
     
                    

var msg = new MyClass{};                            
myBus.Publish(msg, "MyTopic");
     
                    

Request / Response (RPC)

                          
myBus.Respond<MyClass1,MyClass2>(DoSometingWithResult);
myBus.RespondAsync<MyClass1,MyClass2>(DoSometingAsyncWithResult);
                    

var req = new MyClass1{};                            
var result = myBus.Request<MyClass1,MyClass2>(req);
var result = await myBus.RequestAsync<MyClass1,MyClass2>(req);
                    

Send / Receive (Command Pipelines)

myBus.Receive("MyMessageQueue", x => x
    .Add<MyClassA>(DoSomethingWithA)
    .Add<MyClassB>(DoSomethingWithB);
                    

myBus.Send("MyMessageQueue", new MyClassA {});
myBus.Send("MyMessageQueue", new MyClassB {});

                    

Fehlerbehandlung

Weblogger

Prozessketten mit Webbroker

Webbroker / Authorization

• NancyFx empfängt HTTP-Post auf URL http://*:80/service/
• Zugangsdaten werden per Authorization-Service geprüft
• HTTP 200 (OK)
• HTTP 401 (Unauthorized)
• HTTP 500 (JSON gegen Schema nicht valide)
• Weitergabe JSON an Message Router Service

Routing

• Message Router
• Umwandlung JSON in DTO
• Weiterleitung an Section Router Services (Customer / Invoic)
• Customer Router
  • Prüfen des Nachrichtentyps
  • Weiterleitung der Registrierungen an Registration Service
• Invoic Router
  • Prüfen des Nachrichtentyps
  • Weiterleitung der Rechnungen an Bill Service
  • Weiterleitung der Stornierungen an Reversal Service

Customer Services

• Prüfen, ob Kunde bereits vorhanden
• Ja
  • Ablehnung an Absender per Dispatcher senden
• Nein
  • Kunden in der Datenbank erstellen
  • Bestätigung an Absender per Dispatcher senden

INVOIC Services

Message Dispatcher

• DTO in JSON umgewandelt
• Nachricht per HTTP Post an Empfänger versenden

Microservices mit RabbitMQ

Vorteile

• Einfachere Verknüpfung / Routing
• Kompensierung von Lastspitzen
• Kompensierung von Nichtverfügbarkeit
• Skalierung ohne Architekturänderung

Nachteile

• Erhöhte Komplexität (bei kleinem System)
• Implementierung
• Betrieb
• RabbitMQ als SPOF
• Technologiemix schwieriger

EasyNetQ - Erweiterung

Logging mit Log4Net

Prozessketten mit Webbroker für Fortgeschrittene

Problem - große Nachrichten