streamalism.org

1. Overzicht

In dit artikel presenteren we BookKeeper, een service die een gedistribueerd, fouttolerant archiefopslagsysteem.

2. Wat is Boekhouder?

BookKeeper is oorspronkelijk ontwikkeld door Yahoo als een ZooKeeper-subproject en is in 2015 afgestudeerd om een ​​topproject te worden. In de kern wil BookKeeper een betrouwbaar en krachtig systeem zijn dat reeksen van Logboekvermeldingen (ook bekend als Records) in datastructuren genaamd Grootboeken.

Een belangrijk kenmerk van grootboeken is het feit dat ze alleen bijgevoegd en onveranderlijk zijn. Dit maakt BookKeeper een goede kandidaat voor bepaalde toepassingen, zoals gedistribueerde logboeksystemen, Pub-Sub-berichtentoepassingen en realtime streamverwerking.

3. BookKeeper-concepten

3.1. Logboekvermeldingen

Een logboekinvoer bevat een ondeelbare gegevenseenheid die een clienttoepassing opslaat in of leest uit BookKeeper. Bij opslag in een grootboek bevat elk item de aangeleverde gegevens en enkele metadatavelden.

Die metadatavelden bevatten een entryId, die uniek moet zijn binnen een bepaald grootboek. Er is ook een authenticatiecode die BookKeeper gebruikt om te detecteren wanneer een item corrupt is of er mee geknoeid is.

BookKeeper biedt op zichzelf geen serialisatiefuncties, dus klanten moeten hun eigen methode bedenken om constructies op een hoger niveau van / naar te converteren byte arrays.

3.2. Grootboeken

Een grootboek is de basisopslageenheid die wordt beheerd door BookKeeper, waarin een geordende reeks logboekvermeldingen wordt opgeslagen. Zoals eerder vermeld, hebben grootboeken alleen-append-semantiek, wat betekent dat records niet kunnen worden gewijzigd nadat ze eraan zijn toegevoegd.

Ook, zodra een klant stopt met schrijven naar een grootboek en het sluit, BookKeeper zegels en we kunnen er geen gegevens meer aan toevoegen, ook niet op een later tijdstip. Dit is een belangrijk punt om in gedachten te houden bij het ontwerpen van een applicatie rond BookKeeper. Grootboeken zijn geen goede kandidaat om rechtstreeks constructies op een hoger niveau te implementeren, zoals een wachtrij. In plaats daarvan zien we dat grootboeken vaker worden gebruikt om meer basale datastructuren te creëren die deze concepten op een hoger niveau ondersteunen.

Het Distributed Log-project van Apache gebruikt bijvoorbeeld grootboeken als logsegmenten. Die segmenten worden geaggregeerd in gedistribueerde logboeken, maar de onderliggende grootboeken zijn transparant voor reguliere gebruikers.

BookKeeper bereikt grootboekbestendigheid door logboekvermeldingen over meerdere serverinstanties te repliceren. Drie parameters bepalen hoeveel servers en kopieën er worden bewaard:

• Ensemblegrootte: het aantal servers dat wordt gebruikt om grootboekgegevens te schrijven
• Schrijfquorumgrootte: het aantal servers dat wordt gebruikt om een ​​gegeven logboekvermelding te repliceren
• Ack-quorumgrootte: het aantal servers dat een bepaalde schrijfbewerking in het logboekitem moet erkennen

Door die parameters aan te passen, kunnen we de prestatie- en veerkrachtkenmerken van een bepaald grootboek afstemmen. Bij het schrijven naar een grootboek zal BookKeeper de bewerking alleen als succesvol beschouwen wanneer een minimumquorum van clusterleden dit erkent.

Naast de interne metadata, ondersteunt BookKeeper ook het toevoegen van aangepaste metadata aan een grootboek. Dat is een kaart met sleutel / waarde-paren die klanten doorgeven op het moment van creatie en BookKeeper-winkels in ZooKeeper naast zijn eigen.

3.3. Bookies

Bookies zijn servers die één of mode-ledgers bevatten. Een BookKeeper-cluster bestaat uit een aantal bookmakers die in een bepaalde omgeving draaien en diensten verlenen aan klanten via gewone TCP- of TLS-verbindingen.

Bookies coördineren acties met behulp van clusterservices van ZooKeeper. Dit houdt in dat, als we een volledig fouttolerant systeem willen bereiken, we minimaal een ZooKeeper met 3 instances en een BookKeeper-setup met 3 instances nodig hebben. Een dergelijke opstelling zou verlies kunnen tolereren als een enkele instantie faalt en toch normaal kan werken, althans voor de standaard grootboekopstelling: ensemblegrootte met 3 knooppunten, schrijfquorum met 2 knooppunten en 2-knooppuntsquorum.

4. Lokale instellingen

De basisvereisten om BookKeeper lokaal uit te voeren zijn vrij bescheiden. Ten eerste hebben we een ZooKeeper-instantie nodig die actief is en die de opslag van grootboekmetagegevens voor BookKeeper biedt. Vervolgens zetten we een bookmaker in, die de daadwerkelijke diensten aan klanten levert.

Hoewel het zeker mogelijk is om deze stappen handmatig uit te voeren, gebruiken we hier een docker-compose bestand dat officiële Apache-afbeeldingen gebruikt om deze taak te vereenvoudigen:

$ cd $ docker-compose

Dit docker-compose creëert drie bookmakers en een ZooKeeper-instantie. Omdat alle bookmakers op dezelfde machine draaien, is het alleen nuttig voor testdoeleinden. De officiële documentatie bevat de nodige stappen om een ​​volledig fouttolerant cluster te configureren.

Laten we een basistest doen om te controleren of het werkt zoals verwacht, met behulp van de shell-opdracht van de boekhouder listbookies:

$ docker exec -it apache-bookkeeper_bookie_1 / opt / bookkeeper / bin / bookkeeper \ shell listbookies -readwrite ReadWrite Bookies: 192.168.99.101 (192.168.99.101): 4181 192.168.99.101 (192.168.99.101): 4182 192.168.99.101 (192.168. 99.101): 3181 

De uitvoer toont de lijst met beschikbare bookmakers, bestaande uit drie bookmakers. Houd er rekening mee dat de weergegeven IP-adressen zullen veranderen afhankelijk van de specifieke kenmerken van de lokale Docker-installatie.

5. Met behulp van de Ledger API

De Ledger API is de meest eenvoudige manier om te communiceren met BookKeeper. Het stelt ons in staat om rechtstreeks met Grootboek objecten, maar aan de andere kant ontbreekt directe ondersteuning voor abstracties op een hoger niveau, zoals streams. Voor die gevallen biedt het BookKeeper-project een andere bibliotheek, DistributedLog, die deze functies ondersteunt.

Het gebruik van de Ledger API vereist het toevoegen van het boekhouder-server afhankelijkheid van ons project:

 org.apache.bookkeeper boekhouder-server 4.10.0 

OPMERKING: Zoals vermeld in de documentatie, omvat het gebruik van deze afhankelijkheid ook afhankelijkheden voor de protobuf- en guava-bibliotheken. Mocht ons project die bibliotheken ook nodig hebben, maar met een andere versie dan die gebruikt door BookKeeper, dan zouden we een alternatieve afhankelijkheid kunnen gebruiken die die bibliotheken in de schaduw stelt:

 org.apache.bookkeeper bookkeeper-server-shaded 4.10.0 

5.1. Verbinding maken met Bookies

De Boekhouder class is het belangrijkste toegangspunt van de Ledger API, wat een aantal methoden biedt om verbinding te maken met onze BookKeeper-service. In de eenvoudigste vorm hoeven we alleen maar een nieuwe instantie van deze klasse te maken, door het adres van een van de ZooKeeper-servers die door BookKeeper worden gebruikt door te geven:

BookKeeper-client = nieuwe BookKeeper ("zookeeper-host: 2131"); 

Hier, dierenverzorger-host moet worden ingesteld op het IP-adres of de hostnaam van de ZooKeeper-server die de clusterconfiguratie van BookKeeper bevat. In ons geval is dat meestal "localhost" of de host waarnaar de omgevingsvariabele DOCKER_HOST verwijst.

Als we meer controle nodig hebben over de verschillende beschikbare parameters om onze client te verfijnen, kunnen we een Clientconfiguratie instantie en gebruik het om onze klant te maken:

ClientConfiguration cfg = nieuwe ClientConfiguration (); cfg.setMetadataServiceUri ("zk + null: // zookeeper-host: 2131"); // ... stel andere eigenschappen in BookKeeper.forConfig (cfg) .build ();

5.2. Een grootboek aanmaken

Zodra we een Boekhouder Zo is het aanmaken van een nieuw grootboek eenvoudig:

LedgerHandle lh = bk.createLedger (BookKeeper.DigestType.MAC, "wachtwoord" .getBytes ());

Hier hebben we de eenvoudigste variant van deze methode gebruikt. Het maakt een nieuw grootboek met standaardinstellingen, waarbij het MAC-overzichtstype wordt gebruikt om de integriteit van de invoer te garanderen.

Als we aangepaste metadata aan ons grootboek willen toevoegen, moeten we een variant gebruiken die alle parameters accepteert:

LedgerHandle lh = bk.createLedger (3, 2, 2, DigestType.MAC, "wachtwoord" .getBytes (), Collections.singletonMap ("naam", "mijn-grootboek" .getBytes ()));

Deze keer hebben we de volledige versie van het createLedger () methode. De eerste drie argumenten zijn respectievelijk de ensemblegrootte, het schrijfquorum en het ack-quorum. Vervolgens hebben we dezelfde digest-parameters als voorheen. Ten slotte passeren we een Kaart met onze aangepaste metadata.

In beide bovenstaande gevallen createLedger is een synchrone operatie. BookKeeper biedt ook het maken van asynchrone grootboeken met behulp van een callback:

bk.asyncCreateLedger (3, 2, 2, BookKeeper.DigestType.MAC, "passwd" .getBytes (), (rc, lh, ctx) -> {// ... gebruik lh om toegang te krijgen tot grootboekbewerkingen}, null, Collections .emptyMap ()); 

Nieuwere versies van BookKeeper (> = 4.6) ondersteunen ook een vloeiende API en CompletableFuture om hetzelfde doel te bereiken:

CompletableFuture cf = bk.newCreateLedgerOp () .withDigestType (org.apache.bookkeeper.client.api.DigestType.MAC) .withPassword ("wachtwoord" .getBytes ()) .execute (); 

Merk op dat we in dit geval een WriteHandle inplaats van een LedgerHandle. Zoals we later zullen zien, kunnen we ze allemaal gebruiken om toegang te krijgen tot ons grootboek als LedgerHandle werktuigen WriteHandle.

5.3. Gegevens schrijven

Zodra we een LedgerHandle of WriteHandle, schrijven we gegevens naar het bijbehorende grootboek met behulp van een van de toevoegen () methode varianten. Laten we beginnen met de synchrone variant:

for (int i = 0; i <MAX_MESSAGES; i ++) {byte [] data = new String ("message-" + i) .getBytes (); lh.append (data); } 

Hier gebruiken we een variant waarvoor een byte array. De API ondersteunt ook Netty's ByteBuf en Java NIO's ByteBuffer, die een beter geheugenbeheer mogelijk maken in tijdkritische scenario's.

Voor asynchrone bewerkingen verschilt de API een beetje, afhankelijk van het specifieke handle-type dat we hebben verworven. WriteHandle toepassingen Toekomstige, terwijl LedgerHandle ondersteunt ook op callback gebaseerde methoden:

// Beschikbaar in WriteHandle en LedgerHandle CompletableFuture f = lh.appendAsync (data); // Alleen beschikbaar in LedgerHandle lh.asyncAddEntry (data, (rc, ledgerHandle, entryId, ctx) -> {// ... callback-logica weggelaten}, null);

Welke je moet kiezen, is grotendeels een persoonlijke keuze, maar in het algemeen gebruik je CompletableFuture-gebaseerde API's zijn doorgaans gemakkelijker te lezen. Er is ook een bijkomend voordeel dat we een Mono rechtstreeks daaruit, waardoor het gemakkelijker wordt om BookKeeper in reactieve applicaties te integreren.

5.4. Gegevens lezen

Het lezen van gegevens uit een BookKeeper-grootboek werkt op dezelfde manier als schrijven. Ten eerste gebruiken we onze Boekhouder instantie om een LedgerHandle:

LedgerHandle lh = bk.openLedger (ledgerId, BookKeeper.DigestType.MAC, ledgerPassword); 

Behalve de ledgerId parameter, die we later zullen behandelen, deze code lijkt veel op de createLedger () methode die we eerder hebben gezien. Er is echter een belangrijk verschil; deze methode retourneert een alleen-lezen LedgerHandle voorbeeld. Als we een van de beschikbare toevoegen () methoden, alles wat we krijgen is een uitzondering.

Als alternatief is een veiligere manier om de API in vloeiende stijl te gebruiken:

ReadHandle rh = bk.newOpenLedgerOp () .withLedgerId (ledgerId) .withDigestType (DigestType.MAC) .withPassword ("wachtwoord" .getBytes ()) .execute () .get (); 

ReadHandle beschikt over de vereiste methoden om gegevens uit ons grootboek te lezen:

long lastId = lh.readLastConfirmed (); rh.read (0, lastId) .forEach ((entry) -> {// ... doe iets});

Hier hebben we eenvoudig alle beschikbare gegevens in dit grootboek opgevraagd met behulp van de synchrone lezen variant. Zoals verwacht is er ook een asynchrone variant:

rh.readAsync (0, lastId) .thenAccept ((entries) -> {entries.forEach ((entry) -> {// ... procesinvoer});});

Als we ervoor kiezen om het oudere openLedger () methode, zullen we aanvullende methoden vinden die de callback-stijl voor asynchrone methoden ondersteunen:

lh.asyncReadEntries (0, lastId, (rc, lh, entries, ctx) -> {while (entries.hasMoreElements ()) {LedgerEntry e = ee.nextElement ();}}, null);

5.5. Listing Ledgers

We hebben eerder gezien dat we het grootboek nodig hebben ID kaart om de gegevens te openen en te lezen. Dus, hoe krijgen we er een? Een manier is om de LedgerManager interface, waartoe we toegang hebben via onze Boekhouder voorbeeld. Deze interface behandelt in feite metadata van het grootboek, maar heeft ook de asyncProcessLedgers () methode. Met behulp van deze methode - en sommige helpen bij het vormen van gelijktijdige primitieven - kunnen we alle beschikbare grootboeken opsommen:

openbare lijst listAllLedgers (BookKeeper bk) {List ledgers = Collections.synchronizedList (new ArrayList ()); CountDownLatch processDone = nieuwe CountDownLatch (1); bk.getLedgerManager () .asyncProcessLedgers ((ledgerId, cb) -> {ledgers.add (ledgerId); cb.processResult (BKException.Code.OK, null, null);}, (rc, s, obj) -> { processDone.countDown ();}, null, BKException.Code.OK, BKException.Code.ReadException); probeer {processDone.await (1, TimeUnit.MINUTES); terugkeer grootboeken; } catch (InterruptedException ie) {throw nieuwe RuntimeException (ie); }} 

Laten we deze code eens onder de loep nemen, die iets langer is dan verwacht voor een ogenschijnlijk triviale taak. De asyncProcessLedgers () methode vereist twee callbacks.

De eerste verzamelt alle grootboek-id's in een lijst. We gebruiken hier een gesynchroniseerde lijst omdat deze callback kan worden aangeroepen vanuit meerdere threads. Naast het grootboek-ID krijgt deze callback ook een callback-parameter. We moeten haar bellen processResult () methode om te erkennen dat we de gegevens hebben verwerkt en om aan te geven dat we klaar zijn om meer gegevens te ontvangen.

De tweede callback wordt gebeld wanneer alle grootboeken naar de callback van de processor zijn gestuurd of wanneer er een storing is. In ons geval hebben we de foutafhandeling weggelaten. In plaats daarvan verlagen we gewoon een CountDownLatch, die op hun beurt het wachten operatie en laat de methode terugkeren met een lijst van alle beschikbare grootboeken.

6. Conclusie

In dit artikel hebben we het Apache BookKeeper-project besproken, de kernconcepten bekeken en de low-level API gebruikt om toegang te krijgen tot Ledgers en lees- / schrijfbewerkingen uit te voeren.

Zoals gewoonlijk is alle code beschikbaar op GitHub.