distributed-communication.html 15.6 KB
Newer Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
<p>Ein verteiltes System (engl. <i>Distributed System</i>) ist eine Menge unabhängiger Computer, die sich ihren Anwendern als ein einziges zusammenhängendes System präsentieren <a href="#cite-TvS16">[TvS16]</a>. In einem verteilten Anwendungssystem werden also Teilprozesse der Anwendung auf unterschiedlichen Computern ausgeführt.</p>

<div class="cite">"A distributed system is a collection of independent computers that
appears to its users as a single coherent system." (Andrew Tanenbaum)</div>

<p>Ein Computer kann dabei ganz unterschiedlich ausgeprägt sein, z.B. als handelsübliches Notebook oder Smartphone eines Anwenders, als Einplatinencomputer (z.B. <a href="https://raspberrypi.org">Raspberry Pi</a>, <a href="https://arduino.cc">Arduino</a>) eingebettet in ein Gerät mit speziellen Sensoren/Aktuatoren oder als dedizierter oder virtueller Server in einem Rechenzentrum. Es wird i.d.R. eine sogenannte <i><a href="https://de.wikipedia.org/wiki/Shared_Nothing_Architecture">Shared Nothing-Architektur</a></i> angenommen – ein Begriff den Michael Stonebraker bereits 1986 prägte (s. <a href="http://db.cs.berkeley.edu/papers/hpts85-nothing.pdf">The Case for Shared Nothing</a>). Die Computer in einem verteilten System werden dabei vereinfacht Knoten genannt. Jeder Knoten verfügt über mind. einen Prozessor, über flüchtigen Arbeitsspeicher (RAM) und über nicht-flüchtigen Sekundärspeicher (HDD, Flash-Speicher). Die Prozessoren führen die Teilprozesse (engl. <i>Threads</i>) der verteilten Anwendung aus. Da sich die Knoten keinen gemeinsamen Speicher teilen, kommunizieren sie untereinander über Nachrichten. Die Koordination der Knoten erfolgt also auf Softwareebene und setzt eine Netzwerkverbindung der Knoten voraus. Die Netzwerkverbindung kann u.U. vorübergehend unterbrochen sein.</p>

<img src="media/distributed_tanenbaum.png" style="width:500px">
<label>Struktureller Aufbau eines verteilten Systems nach <a href="#cite-TvS16">[TvS16]</a></label>

<p>In der obigen Abbildung ist zu erkennen, dass alle am System beteiligten Computer mit unterschiedlichen Betriebssystemen (engl. <i>Operating System</i>, kurz <i>OS</i>) ausgestattet sein können. Die Komponenten einer verteilten Anwendung können direkt untereinander kommunizieren – wie für Application B in der obigen Abbildung angedeutet. In den folgenden Kapiteln werden wir aktuell geläufige Protokolle für diese Interprozesskommunikation, die die Grenzen eines Computers überwindet, vorstellen, z.B. <a href="#unit-rmi" class="navigate">RMI</a>, <a href="#unit-soap" class="navigate">SOAP</a> und <a href="#unit-websocket" class="navigate">WebSockets</a>. Alternativ können die Komponenten in einem verteilten System auch indirekt Nachrichten über eine vermittelnde Middleware austauschen. Neben verteilten Datenbanksystemen bieten sich hier insbesondere zuverlässige <i>Message Broker</i> wie <a href="https://kafka.apache.org/">Kafka</a> oder <a href="https://www.rabbitmq.com/">RabbitMQ</a> an, die auf einen hohen Durchsatz an vermittelten Nachrichten ausgelegt sind.</p>

<p>Eine <i>Shared Nothing-Architektur</i> lässt sich leicht skalieren, indem weitere Knoten dem System hinzugefügt werden oder wieder entfernt werden. Die maßgeblichen Motivationsfaktoren bei der Skalierung sind eine Verbesserung der Performance und eine Verbesserung der Datensicherheit durch Replikation.</p>

<h4>Kommunikation im verteilten System</h4>

<p>Es existiert eine Vielzahl von Protokollen zur Kommunikation in einem verteilten System, d.h. zum Austausch von Nachrichten/Daten zwischen Software-Komponenten, die in Prozessen auf verschiedenen Computern ausgeführt werden und über ein Netzwerk miteinander verbunden sind. Entsprechend den Anforderungen eines konkreten Anwendungsfalls ist ein geeignetes Protokoll als Grundlage für diese Interprozesskommunikation auszuwählen. Eine erste Orientierung bietet das <a href="https://de.wikipedia.org/wiki/Internetprotokollfamilie">TCP/IP-Modell</a>, das die Kommunikationsprotokolle gemäß der folgenden Abbildung in 4 Schichten untergliedert.</p>

<img src="media/distributed_tcp_ip_model.png" style="width:900px">
20
<label>Protokolle zur Remote-Kommunikation und das TCP/IP-Schichtenmodell</label>
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

<p>Aus der Perspektive eines Anwendungsentwicklers möchten wir uns auf die Protokolle der unteren Schichten (Transport-, Internet-, Netzzugangsschicht im TCP/IP-Modell) verlassen und ein adäquates Protokoll auf Ebene der Anwendungsschicht nutzen. Wir verfolgen grundsätzlich das Ziel, dass eine lokale Client-Komponente eine Kommunikationsverbindung zu einer entfernten (engl. <i>remote</i>) Server-Komponente aufbaut und anschließend anwendungsspezifische Daten austauschen kann. In den nächsten Kapiteln werden dazu folgende alternative Ansätze zur sogenannten Remote-Kommunikation am Beispiel von Java-Anwendungen vorgestellt:</p>
<ul>
<li><a href="#unit-rmi" class="navigate">Remote Method Invocation (RMI)</a></li>
<li><a href="#unit-websocket" class="navigate">WebSockets</a></li>
<li><a href="#unit-socket" class="navigate">Sockets</a></li>
<li><a href="#unit-soap" class="navigate">SOAP-Webservices</a></li>
<li><a href="#unit-rest" class="navigate">REST-Webservices</a></li>
</ul>

<span>Eine wichtige Entwurfsentscheidung ist, ob die Kommunikation zwischen dem aufrufenden Client und dem Server nach dem <i>Pull-Prinzip</i> oder nach dem <i>Push-Prinzip</i> erfolgen soll. Der Unterschied zwischen beiden Kommunikationsprinzipien ist bereits im Kapitel zum <a href="">Beobachter-Muster</a> vorgestellt worden.</span>
<ul>
<li><b>Pull-Prinzip</b>: Der Client erfragt den aktuellen Status einer Ressource beim Server. RMI und insbesondere Webservices sind konzeptionell auf das Pull-Prinzip ausgerichtet. Da sowohl SOAP- als auch REST-basierte Webservices auf dem HTTP-Protokoll aufsetzen, ist nur eine unidirektionale Kommunikation möglich.</li>
<li><b>Push-Prinzip</b>: Der Server teilt den vorab registrierten Clients kontinuierlich Statusänderungen mit. Sockets und WebSockets sind konzeptionell auf das Push-Prinzip ausgerichtet. Auch bei RMI lässt sich das Beobachter-Muster umsetzen, um eine derartige bidirektionale Kommunikation abzubilden.</li>
</ul>

<span>Eine weitere Entwurfsentscheidung ist, ob die Remote-Kommunikation zwischen Client und Server synchron oder asynchron erfolgen soll.</span>
<ul>
<li><b>Synchrone Kommunikation</b>: Der Client wartet nach einer Anfrage an den Server auf dessen Antwort und ist währenddessen blockiert. Erst nach dem Empfang der Antwort wird der Kontrollfluss im Client fortgesetzt.</li>
<li><b>Asynchrone Kommunikation</b>: Der Client sendet eine Anfrage an den Server und setzt anschließend den Kontrollfluss – ohne zu blockieren – unmittelbar fort. Wenn die Antwort des Servers später irgendwann eingeht, wird sie mittels einer Callback-Funktion verarbeitet.</li>
</ul>
<span>Es ist allgemein anzustreben, dass die Remote-Kommunikation (z.B. in Form von HTTP-Requests) nicht das User Interface eines Client blockiert. Insbesondere bei synchroner Remote-Kommunikation ist darauf zu achten, dass die Anfragen an einen Server nicht aus dem Thread der Anwendung abgesetzt werden, in dem auch die Anwenderinteraktion verarbeitet wird. Auf der Serverseite ist zu berücksichtigen, dass i.d.R. mit mehreren Clients parallel kommuniziert werden soll. Daher ist sowohl bei der Implementierung der Client-Komponente als auch bei der Implementierung der Server-Komponente zu empfehlen, die Remote-Kommunikation in nebenläufige Threads zu verlagern.</span>

<h4>Performance</h4>

<p>Die Performance eines Anwendungssystems beschreibt den Erfüllungsgrad von spezifizierten Zielen bezüglich der Metriken Antwortzeit und Durchsatz <a href="#cite-SW01">[SW01]</a>. Beide Metriken lassen sich einfach beobachten. Die Antwortzeit (engl. <i>Response Time</i>) wird i.d.R. aus Sicht der Anwenders gemessen, der eine Anfrage an das System sendet, und kann wie in der folgenden Abbildung dargestellt in Netzwerkzeit, Wartezeit und Bedienzeit zerlegt werden. Jeder dieser Teilaspekte der Antwortzeit kann für sich optimiert werden.</p>

<img src="media/distributed_response_time.png" style="width:640px">
<label>Unterteilung der Antwortzeit eines Systems in Netzwerk-, Warte- und Bedienzeit</label>

<div class="cite">"Performance is the degree to which a software system or component meets its objectives for timeliness. There are two important dimensions to software performance: responsiveness and scalability. Responsiveness is the ability of a system to meet its objectives for response time or throughput. Scalability is the ability of a system to continue to meet its response time or throughput objectives as the demand for the software functions increases." (Connie Smith, Lloyd Williams)</div>

<p>Es ergeben sich folgende Einflussfaktoren auf die Performance einer Komponente zur Laufzeit:</p>

<img src="media/distributed_performance_drivers.png" style="width:600px">
56
<label>Einflussfaktoren auf die Antwortzeit einer Komponente</label>
57
58
59
60
61
62
63
64
65
66
67
68

<ul>
<li><b>Implementierung</b>: In Abhängigkeit der verwendeten Datenstrukturen und Algorithmen kann eine Komponente unterschiedlich effizient implementiert werden.</li>
<li><b>Deployment</b>: Die Laufzeitumgebung beeinflusst die Performance einer Komponente maßgeblich. Zur Laufzeitumgebung zählen neben den eigentlichen Hardware-Ressourcen auch das Betriebssystem und andere Software-Container. Der einfachste Ansatz die Performance zu verbessern, ist das Prinzip <i>"Kill it with iron"</i>, d.h. leistungsfähigere Hardware-Ressourcen einzusetzen.</li>
<li><b>Abhängigkeiten</b>: Wenn externe Dienste anderer Komponenten aus einer Komponente heraus (insbesondere synchron) aufgerufen werden, wird dadurch die Antwortzeit beeinflusst – z.B. bei Anfragen an Datenbanksysteme oder Webservices. Die Abhängigkeiten einer Komponenten sind i.d.R. vor dem Aufrufer verborgen. Dadurch kann der Aufrufer bei einer Störung nicht unterscheiden, ob die Komponente selbst oder einer der Dienste, von denen sie abhängt, die Störung verursachen.</li>
<li><b>Anwenderverhalten</b>: Die Performance einer Komponente wird vom Interaktionsverhalten der Anwender beeinflusst. Antwortzeiten und Durchsatz hängen vom aktuellen <i>Workload</i> ab, d.h. in welcher Frequenz neue Anfragen eingehen und wie viel Datentransfer und Berechnungsaufwand durch die unterschiedliche Parametrisierung der Anfragen verursacht wird. Eine Komponente sollte möglichst mehrere Anfragen parallel verarbeiten können. Es kann notwendig sein, Prioritäten beim Scheduling der Anfragen zu berücksichtigen.</li>
</ul>
</p>

<p>Skalierbarkeit drückt das Verhältnis zwischen der Steigerung der Problemgröße (z.B. mehr Workload) und den zur Lösung zusätzlich benötigten Ressourcen aus. Antwortzeiten und Durchsatz skalieren bei steigendem Workload und beschränkter Ressourcen-Kapazität bis zum Erreichen der Kapazitätsgrenze etwa linear. Anschließend steigen die Antwortzeiten exponentiell, da eine bestimmte Ressource (z.B. CPU, RAM, I/O) aufgrund von Überlastung zum Engpass wird. Dieser typische Effekt ist in der folgenden Abbildung dargestellt. Gleichermaßen erreicht auch der Durchsatz einen Sättigungspunkt, an dem die Komponente oder das System einfach nicht mehr Anfragen pro Zeitintervall verarbeiten kann. In automatisch skalierenden Cloud-Plattformen kann dieser Sättigungspunkt soweit hinausgezögert werden, das er praktisch nicht erreicht wird. Allgemein ist es aus Gründen der Wirtschaftlichkeit zu empfehlen, die bereitgestellten Hardware-Ressourcen nicht zu stark überzudimensionieren.</p>

<img src="media/distributed_scalability.png" style="width:600px">
69
<label>Skalierbarkeit von Antwortzeit und Durchsatz bei beschränkter Ressourcen-Kapazität</label>
70
71
72
73

<p>Um die Ressourcen-Kapazität eines Systems auszubauen, gibt es zwei grundlegende Ansätze: die vertikale Skalierung (<i>Scale Up</i>) und die horizontale Skalierung (<i>Scale Out</i>). Bei der vertikalen Skalierung wird ein leistungsstarker Großrechner angestrebt, der sukzessiv ausgebaut wird. Dieser Ansatz gilt als vergleichsweise teuer. Außerdem stellt der Großrechner einen möglichen <i>Single Point of Failure</i> dar. Daher wird heute horizontale Skalierung durch ein verteiltes System von vielen gleichartigen Knoten favorisiert. In einem derartigen <i>Cluster</i> wird der Ausfall von wenigen Knoten als Normalzustand angenommen, der die Verfügbarkeit und die Konsistenz der Daten nicht gefährden darf. Zwar ist ein Cluster komplexer zu beherrschen als ein einzelner Großrechner, kann dafür aber skalierbare Performance und Datensicherheit gleichermaßen gewährleisten.</p>

<img src="media/distributed_scale_out.png" style="width:700px">
74
<label>Vertikale und horizontale Skalierung</label>
75
76
77
78
79

<h4>Datensicherheit durch Replikation</h4>
<p>Nehmen wir an, dass wie oben dargestellt ein Cluster von gleichartigen Knoten aufgebaut worden ist, um auch bei hohem Workload die geforderte Performance des Systems zu gewährleisten. Eine Komponente wird innerhalb des Clusters nun mehrfach instantiiert, damit der Dienst, den sie anbietet, ebenso skaliert. Jede Instanz dieser Komponente wird nur einen Ausschnitt der gesamten Datenbasis verarbeiten und auf dem Knoten, auf dem sie ausgeführt wird, speichern. Die Daten werden in einem verteilten Datei- oder Datenbanksystem in sogenannte Partitionen (oder auch <i>Shards</i>) zerlegt. Einzelne Datenobjekte/Datensätze werden i.d.R. über eine Hashfunktion einer Partition zugeordnet. Die folgende Abbildung zeigt ein Cluster mit 4 Knoten, auf denen jeweils eine Instanz der gleichen Komponente ausgeführt wird. Die Datenbasis wurde in 6 Partitionen (A-F) unterteilt. Da das System keine Daten verlieren darf, wenn ein Knoten ausfällt, werden die Partitionen repliziert. Im dargestellten Beispiel ist der Replikationsfaktor 3, d.h. von jeder Partition gibt 3 identische Replikate, so dass das abgebildete System den Ausfall beliebiger 2 Knoten tolerieren kann. Die Anzahl der Knoten und insbesondere der Partitionen ist in der Praxis häufig deutlich größer. Eine größere Anzahl an Partitionen als Knoten vereinfacht die Umverteilung der Partitionen, wenn ein neuer Knoten in den Cluster aufgenommen wird oder ein Knoten ausscheidet.</p>

<img src="media/distributed_replication.png" style="width:700px">
80
<label>Partitionierung und Replikation von Daten im verteilten System</label>
81
82
83

<p>Ein Kompromiss zwischen Verfügbarkeit und Konsistenz der Daten ergibt sich dadurch, dass die Replikate nach einem Schreibzugriff synchronisiert werden müssen. Es stellt sich in einem verteilten System die Frage, ob während der Synchronisation bedingt durch einen Schreibzugriff entweder ein veralteter Zustand gelesen werden darf (d.h. Verfügbarkeit nicht eingeschränkt, aber Konsistenz eingeschränkt) oder das Lesen bis zum Ende der Synchronisation verzögert wird (d.h. Konsistenz nicht eingeschränkt, aber Verfügbarkeit eingeschränkt). In einem verteilten System ist grundsätzlich mit dem Ausfall einzelner Knoten oder der vorübergehenden Unterbrechung der Netzwerkverbindungen zu rechnen, was zu Verzögerungen in der Synchronisation führt. Der Begriff Partitionstoleranz beschreibt, dass das System bis zu einem gewissen Grad tolerant gegenüber einem kurzzeitigen Verfall des Systems in unverbundene Teile sein muss. Der Kompromiss zwischen der Konsistenz (engl. <i><u>C</u>onsistency</i>), der Verfügbarkeit (engl. <i><u>A</u>vailability</i>) und der Partitionstoleranz (engl. <i><u>P</u>artition Tolerance</i>) in einem verteilten System ist als <a href="https://en.wikipedia.org/wiki/CAP_theorem">CAP-Theorem</a> bekannt. Verteilte Datenbanken werden in diesem Model nicht weiter behandelt. Einen fundierten Überblick liefert das populäre Buch "Designing Data-Intensive Applications" von Martin Kleppmann <a href="#cite-Kle17">[Kle17]</a>.