streamalism.org

1. Overzicht

In dit artikel zullen we kijken naar conflictvrije gerepliceerde gegevenstypen (CRDT) en hoe u ermee kunt werken in Java. Voor onze voorbeelden gebruiken we implementaties van de wurmloch-crdt bibliotheek.

Als we een cluster hebben van N replicaknooppunten in een gedistribueerd systeem, kunnen we een netwerkpartitie - sommige knooppunten kunnen tijdelijk niet met elkaar communiceren. Deze situatie wordt een gespleten brein genoemd.

Als we een gespleten brein in ons systeem hebben, sommige schrijfverzoeken - zelfs voor dezelfde gebruiker - kunnen naar verschillende replica's gaan die niet met elkaar zijn verbonden. Wanneer een dergelijke situatie zich voordoet, onze systeem is nog steeds beschikbaar, maar is niet consistent.

We moeten beslissen wat we doen met schrijfbewerkingen en gegevens die niet consistent zijn wanneer het netwerk tussen twee gesplitste clusters weer begint te werken.

2. Conflictvrije gerepliceerde gegevenstypen aan de redding

Laten we eens kijken naar twee knooppunten, EEN en B, die zijn losgekoppeld als gevolg van een gespleten brein.

Laten we zeggen dat een gebruiker zijn login verandert en dat een verzoek naar het knooppunt gaat EEN. Vervolgens besluit hij / zij het opnieuw te wijzigen, maar deze keer gaat het verzoek naar het knooppunt B.

Vanwege het gespleten brein zijn de twee knooppunten niet met elkaar verbonden. We moeten beslissen hoe de login van deze gebruiker eruit moet zien als het netwerk weer werkt.

We kunnen een aantal strategieën gebruiken: we kunnen de gebruiker de mogelijkheid geven om conflicten op te lossen (zoals wordt gedaan in Google Documenten), of wij kunnengebruik een CRDT voor het samenvoegen van gegevens van uiteenlopende replica's voor ons.

3. Maven Afhankelijkheid

Laten we eerst een afhankelijkheid aan de bibliotheek toevoegen die een reeks nuttige CRDT's biedt:

 com.netopyr.wurmloch wurmloch-crdt 0.1.0 

De nieuwste versie is te vinden op Maven Central.

4. Alleen kweekset

De meest elementaire CRDT is een alleen-kweekset. Elementen kunnen alleen worden toegevoegd aan een GSet en nooit verwijderd. Wanneer de GSet divergeert, het kan zijn gemakkelijk samengevoegd door de unie te berekenen van twee sets.

Laten we eerst twee replica's maken om een ​​gedistribueerde gegevensstructuur te simuleren en die twee replica's met elkaar verbinden met behulp van de aansluiten() methode:

LocalCrdtStore crdtStore1 = nieuwe LocalCrdtStore (); LocalCrdtStore crdtStore2 = nieuwe LocalCrdtStore (); crdtStore1.connect (crdtStore2);

Zodra we twee replica's in ons cluster hebben, kunnen we een GSet op de eerste replica en verwijs ernaar op de tweede replica:

GSet replica1 = crdtStore1.createGSet ("ID_1"); GSet replica2 = crdtStore2.findGSet ("ID_1"). Get ();

Op dit punt werkt ons cluster zoals verwacht en is er een actieve verbinding tussen twee replica's. We kunnen twee elementen van twee verschillende replica's aan de set toevoegen en stellen dat de set dezelfde elementen op beide replica's bevat:

replica1.add ("appel"); replica2.add ("banaan"); assertThat (replica1) .contains ("appel", "banaan"); assertThat (replica2) .contains ("appel", "banaan");

Laten we zeggen dat we ineens een netwerkpartitie hebben en dat er geen verbinding is tussen de eerste en tweede replica's. We kunnen de netwerkpartitie simuleren met behulp van de verbinding verbreken() methode:

crdtStore1.disconnect (crdtStore2);

Als we vervolgens elementen aan de dataset van beide replica's toevoegen, zijn die wijzigingen niet globaal zichtbaar omdat er geen verband tussen is:

replica1.add ("aardbei"); replica2.add ("peer"); assertThat (replica1) .contains ("appel", "banaan", "aardbei"); assertThat (replica2) .contains ("appel", "banaan", "peer");

Zodra de verbinding tussen beide clusterleden weer tot stand is gebracht, wordt het GSet is samengevoegd intern gebruikmakend van een unie op beide sets, en beide replica's zijn weer consistent:

crdtStore1.connect (crdtStore2); assertThat (replica1) .contains ("appel", "banaan", "aardbei", "peer"); assertThat (replica2) .contains ("appel", "banaan", "aardbei", "peer");

5. Alleen incrementele teller

Alleen incrementteller is een CRDT die alle incrementen lokaal op elk knooppunt aggregeert.

Wanneer replica's worden gesynchroniseerd, wordt na een netwerkpartitie de resulterende waarde berekend door alle verhogingen op alle knooppunten bij elkaar op te tellen - dit is vergelijkbaar met LongAdder van java.concurrent maar op een hoger abstractieniveau.

Laten we een teller voor alleen incrementen maken met GCounter en verhoog het vanaf beide replica's. We kunnen zien dat de som correct wordt berekend:

LocalCrdtStore crdtStore1 = nieuwe LocalCrdtStore (); LocalCrdtStore crdtStore2 = nieuwe LocalCrdtStore (); crdtStore1.connect (crdtStore2); GCounter replica1 = crdtStore1.createGCounter ("ID_1"); GCounter replica2 = crdtStore2.findGCounter ("ID_1"). Get (); replica1.increment (); replica2.increment (2L); assertThat (replica1.get ()). isEqualTo (3L); assertThat (replica2.get ()). isEqualTo (3L); 

Wanneer we beide clusterleden loskoppelen en lokale increment-bewerkingen uitvoeren, kunnen we zien dat de waarden inconsistent zijn:

crdtStore1.disconnect (crdtStore2); replica1.increment (3L); replica2.increment (5L); assertThat (replica1.get ()). isEqualTo (6L); assertThat (replica2.get ()). isEqualTo (8L);

Maar zodra het cluster weer gezond is, worden de incrementen samengevoegd, wat de juiste waarde oplevert:

crdtStore1.connect (crdtStore2); assertThat (replica1.get ()) .isEqualTo (11L); assertThat (replica2.get ()) .isEqualTo (11L);

6. PN-teller

Door een vergelijkbare regel te gebruiken voor de teller voor alleen incrementen, kunnen we een teller maken die zowel kan worden verhoogd als verlaagd. De PNCounter slaat alle verhogingen en verlagingen afzonderlijk op.

Wanneer replica's worden gesynchroniseerd, is de resulterende waarde gelijk aan desom van alle verhogingen minus de som van alle verlagingen:

@Test openbare leegte gegevenPNCounter_whenReplicasDiverge_thenMergesWithoutConflict () {LocalCrdtStore crdtStore1 = nieuwe LocalCrdtStore (); LocalCrdtStore crdtStore2 = nieuwe LocalCrdtStore (); crdtStore1.connect (crdtStore2); PNCounter replica1 = crdtStore1.createPNCounter ("ID_1"); PNCounter replica2 = crdtStore2.findPNCounter ("ID_1"). Get (); replica1.increment (); replica2.decrement (2L); assertThat (replica1.get ()). isEqualTo (-1L); assertThat (replica2.get ()). isEqualTo (-1L); crdtStore1.disconnect (crdtStore2); replica1.decrement (3L); replica2.increment (5L); assertThat (replica1.get ()). isEqualTo (-4L); assertThat (replica2.get ()). isEqualTo (4L); crdtStore1.connect (crdtStore2); assertThat (replica1.get ()). isEqualTo (1L); assertThat (replica2.get ()). isEqualTo (1L); }

7. Last-Writer-Wins-register

Soms hebben we complexere bedrijfsregels en is het niet voldoende om op sets of tellers te werken. We kunnen het Last-Writer-Wins-register gebruiken, dat behoudt alleen de laatst bijgewerkte waarde bij het samenvoegen van uiteenlopende gegevenssets. Cassandra gebruikt deze strategie om conflicten op te lossen.

We moeten Wees zeer voorzichtig bij het gebruik van deze strategie, omdat het veranderingen laat vallen die in de tussentijd zijn opgetreden.

Laten we een cluster maken van twee replica's en instanties van het LWWRegister klasse:

LocalCrdtStore crdtStore1 = nieuwe LocalCrdtStore ("N_1"); LocalCrdtStore crdtStore2 = nieuwe LocalCrdtStore ("N_2"); crdtStore1.connect (crdtStore2); LWWRegister replica1 = crdtStore1.createLWWRegister ("ID_1"); LWWRegister replica2 = crdtStore2.findLWWRegister ("ID_1"). Get (); replica1.set ("appel"); replica2.set ("banaan"); assertThat (replica1.get ()). isEqualTo ("banaan"); assertThat (replica2.get ()). isEqualTo ("banaan"); 

Wanneer de eerste replica de waarde instelt op appel en de tweede verandert het in banaan, de LWWRegister behoudt alleen de laatste waarde.

Laten we eens kijken wat er gebeurt als het cluster de verbinding verbreekt:

crdtStore1.disconnect (crdtStore2); replica1.set ("aardbei"); replica2.set ("peer"); assertThat (replica1.get ()). isEqualTo ("aardbei"); assertThat (replica2.get ()). isEqualTo ("peer");

Elke replica bewaart zijn lokale kopie van gegevens die inconsistent zijn. Als we de set () methode, de LWWRegister wijst intern een speciale versiewaarde toe die de specifieke update identificeert aan elke gebruiker van een VectorKlok algoritme.

Wanneer het cluster synchroniseert, it neemt de waarde met de hoogste versieenverwijdert elke vorige update:

crdtStore1.connect (crdtStore2); assertThat (replica1.get ()). isEqualTo ("peer"); assertThat (replica2.get ()). isEqualTo ("peer");

8. Conclusie

In dit artikel hebben we het probleem van de consistentie van gedistribueerde systemen met behoud van beschikbaarheid laten zien.

In het geval van netwerkpartities moeten we de divergerende gegevens samenvoegen wanneer het cluster wordt gesynchroniseerd. We hebben gezien hoe CRDT's kunnen worden gebruikt om uiteenlopende gegevens samen te voegen.

Al deze voorbeelden en codefragmenten zijn te vinden in het GitHub-project - dit is een Maven-project, dus het moet gemakkelijk te importeren en uit te voeren zijn zoals het is.