streamalism.org

1. Overzicht

In deze zelfstudie leren we hoe u een Ring Buffer in Java implementeert.

2. Ringbuffer

Ring Buffer (of Circular Buffer) is een begrensde circulaire gegevensstructuur die wordt gebruikt voor het bufferen van gegevens tussen twee of meer threads. Terwijl we naar een ringbuffer blijven schrijven, wikkelt het zich rond als het einde bereikt.

2.1. Hoe het werkt

Een ringbuffer wordt geïmplementeerd met behulp van een array met een vaste grootte die zich om de grenzen heen wikkelt.

Afgezien van de array houdt het drie dingen bij:

• het volgende beschikbare slot in de buffer om een ​​element in te voegen,
• het volgende ongelezen element in de buffer,
• en het einde van de array - het punt waarop de buffer rond het begin van de array loopt

De mechanica van hoe een ringbuffer met deze vereisten omgaat, varieert met de implementatie. Het Wikipedia-artikel over het onderwerp toont bijvoorbeeld een methode met vierpunters.

We lenen de aanpak van Disruptors implementatie van de ringbuffer met behulp van sequenties.

Het eerste dat we moeten weten, is de capaciteit - de vaste maximale grootte van de buffer. De volgende, we zullen er twee gebruiken die monotoon toenemenopeenvolgingen:

• Schrijfreeks: beginnend bij -1, wordt met 1 verhoogd wanneer we een element invoegen
• Leesvolgorde: beginnend bij 0, wordt met 1 verhoogd naarmate we een element consumeren

We kunnen een reeks toewijzen aan een index in de array door een mod-bewerking te gebruiken:

arrayIndex = sequentie% capaciteit 

De mod-bewerking wikkelt de reeks rond de grenzen om een ​​slot in de buffer af te leiden:

Laten we eens kijken hoe we een element zouden invoegen:

buffer [++ writeSequence% capaciteit] = element 

We verhogen de reeks vooraf voordat we een element invoegen.

Om een ​​element te consumeren, doen we een post-increment:

element = buffer [readSequence ++% capaciteit] 

In dit geval voeren we een post-increment uit op de reeks. Als je een element consumeert, wordt het niet uit de buffer verwijderd - het blijft gewoon in de array totdat het wordt overschreven.

2.2. Lege en volle buffers

Terwijl we om de array heen lopen, zullen we beginnen met het overschrijven van de gegevens in de buffer. Als de buffer vol is, kunnen we ervoor kiezen om de oudste gegevens te overschrijven, ongeacht of de lezer deze heeft gebruikt, of om te voorkomen dat de gegevens die niet zijn gelezen, worden overschreven.

Als de lezer het zich kan veroorloven om de tussenliggende of oude waarden te missen (bijvoorbeeld een aandelenkoers), kunnen we de gegevens overschrijven zonder te wachten tot ze zijn verbruikt. Aan de andere kant, als de lezer alle waarden moet consumeren (zoals bij e-commercetransacties), moeten we wachten (blok / bezet-wachten) totdat de buffer een slot beschikbaar heeft.

De buffer is vol als de grootte van de buffer gelijk is aan zijn capaciteit, waarbij de grootte gelijk is aan het aantal ongelezen elementen:

size = (writeSequence - readSequence) + 1 isFull = (size == capaciteit) 

Als de schrijfvolgorde achterblijft bij de leesvolgorde, is de buffer leeg:

isEmpty = writeSequence <readSequence 

De buffer retourneert een nul waarde als het leeg is.

2.2. Voor-en nadelen

Een ringbuffer is een efficiënte FIFO-buffer. Het maakt gebruik van een array met een vaste grootte die vooraf kan worden toegewezen en zorgt voor een efficiënt geheugentoegangspatroon. Alle bufferbewerkingen zijn in constante tijd O (1), inclusief het consumeren van een element, aangezien het geen verschuiving van elementen vereist.

Aan de andere kant is het van cruciaal belang om de juiste grootte van de ringbuffer te bepalen. De schrijfbewerkingen kunnen bijvoorbeeld gedurende een lange tijd worden geblokkeerd als de buffer te klein is en het lezen traag is. We kunnen dynamische dimensionering gebruiken, maar daarvoor zouden gegevens moeten worden verplaatst en zullen we de meeste van de hierboven besproken voordelen missen.

3. Implementatie in Java

Nu we begrijpen hoe een ringbuffer werkt, gaan we deze in Java implementeren.

3.1. Initialisatie

Laten we eerst een constructor definiëren die de buffer initialiseert met een vooraf gedefinieerde capaciteit:

openbare CircularBuffer (int capaciteit) {this.capacity = capaciteit; this.data = (E []) nieuw object [capaciteit]; this.readSequence = 0; this.writeSequence = -1; } 

Dit zal een lege buffer creëren en de sequentievelden initialiseren zoals besproken in de vorige sectie.

3.3. Aanbod

Vervolgens implementeren we het aanbod bewerking die een element in de buffer invoegt bij het volgende beschikbare slot en terugkeert waar op succes. Het keert terug false als de buffer geen leeg slot kan vinden, dat wil zeggen, we kunnen geen ongelezen waarden overschrijven.

Laten we het aanbod methode in Java:

openbare booleaanse aanbieding (E-element) {boolean isFull = (writeSequence - readSequence) + 1 == capaciteit; if (! isFull) {int nextWriteSeq = writeSequence + 1; data [nextWriteSeq% capaciteit] = element; writeSequence ++; terugkeer waar; } return false; } 

Dus we verhogen de schrijfvolgorde en berekenen de index in de array voor het volgende beschikbare slot. Vervolgens schrijven we de gegevens naar de buffer en slaan we de bijgewerkte schrijfvolgorde op.

Laten we het uitproberen:

@Test openbare leegte gegevenCircularBuffer_whenAnElementIsEnqueued_thenSizeIsOne () {CircularBuffer buffer = nieuwe CircularBuffer (defaultCapacity); assertTrue (buffer.offer ("Square")); assertEquals (1, buffer.size ()); } 

3.4. Poll

Ten slotte implementeren we het poll bewerking die het volgende ongelezen element ophaalt en verwijdert. De poll operatie verwijdert het element niet, maar verhoogt de leesvolgorde.

Laten we het implementeren:

openbare E poll () {boolean isEmpty = writeSequence <readSequence; if (! isEmpty) {E nextValue = data [readSequence% capaciteit]; readSequence ++; return nextValue; } retourneer null; } 

Hier lezen we de gegevens in de huidige leesvolgorde door de index in de array te berekenen. Vervolgens verhogen we de reeks en retourneren we de waarde als de buffer niet leeg is.

Laten we het uittesten:

@Test openbare leegte gegevenCircularBuffer_whenAnElementIsDequeued_thenElementMatchesEnqueuedElement () {CircularBuffer buffer = nieuwe CircularBuffer (defaultCapacity); buffer.offer ("Driehoek"); Tekenreeksvorm = buffer.poll (); assertEquals ("Triangle", vorm); } 

4. Producent-consumentprobleem

We hebben het gehad over het gebruik van een ringbuffer voor het uitwisselen van gegevens tussen twee of meer threads, wat een voorbeeld is van een synchronisatieprobleem dat het Producer-Consumer-probleem wordt genoemd. In Java kunnen we het producent-consumentprobleem op verschillende manieren oplossen met behulp van semaforen, begrensde wachtrijen, ringbuffers, enz.

Laten we een oplossing implementeren op basis van een ringbuffer.

4.1. vluchtig Volgorde Velden

Onze implementatie van de ringbuffer is niet thread-safe. Laten we het thread-safe maken voor de eenvoudige case voor één producent en één consument.

De producent schrijft gegevens naar de buffer en verhoogt de writeSequence, terwijl de consument alleen uit de buffer leest en de readSequence. De achtergrondarray is dus vrij van conflicten en we kunnen wegkomen zonder enige synchronisatie.

Maar we moeten er nog steeds voor zorgen dat de consument de laatste waarde van de writeSequence veld (zichtbaarheid) en dat de writeSequence wordt niet bijgewerkt voordat de gegevens daadwerkelijk beschikbaar zijn in de buffer (bestellen).

We kunnen de ringbuffer in dit geval gelijktijdig en lock-free maken door de sequentievelden te maken vluchtig:

privé vluchtige int writeSequence = -1, readSequence = 0; 

In de aanbod methode, een schrijven naar de vluchtig veld- writeSequence garandeert dat het schrijven naar de buffer plaatsvindt voordat de volgorde wordt bijgewerkt. Tegelijkertijd is het vluchtig zichtbaarheid garantie zorgt ervoor dat de consument altijd de laatste waarde van ziet writeSequence.

4.2. Producent

Laten we een eenvoudige producer implementeren Runnable die naar de ringbuffer schrijft:

public void run () {for (int i = 0; i <items.length;) {if (buffer.offer (items [i])) {System.out.println ("Produced:" + items [i]) ; i ++; }}} 

De producer-thread zou wachten op een leeg slot in een lus (bezig met wachten).

4.3. Klant

We implementeren een consument Oproepbaar dat leest uit de buffer:

openbare T [] call () {T [] items = (T []) nieuw object [verwachtCount]; voor (int i = 0; i <items.length;) {T item = buffer.poll (); if (item! = null) {items [i ++] = item; System.out.println ("Consumed:" + item); } } items terugbrengen; } 

De consumententhread gaat door zonder af te drukken als deze een nul waarde uit de buffer.

Laten we onze stuurprogrammacode schrijven:

executorService.submit (nieuwe Thread (nieuwe Producer (buffer))); executorService.submit (nieuwe thread (nieuwe consument (buffer))); 

Het uitvoeren van ons producer-consumer-programma produceert output zoals hieronder:

Geproduceerd: cirkel geproduceerd: driehoek verbruikt: cirkel geproduceerd: rechthoek verbruikt: driehoek verbruikt: rechthoek geproduceerd: vierkant geproduceerd: ruit verbruikt: vierkant geproduceerd: trapezium verbruikt: ruit verbruikt: trapezium geproduceerd: vijfhoek geproduceerd: pentagram geproduceerd: zeshoek verbruikt: vijfhoek verbruikt: Pentagram geproduceerd: hexagram verbruikt: zeshoek verbruikt: hexagram 

5. Conclusie

In deze tutorial hebben we geleerd hoe we een ringbuffer kunnen implementeren en hebben we onderzocht hoe deze kan worden gebruikt om het probleem van producent en consument op te lossen.

Zoals gewoonlijk is de broncode voor alle voorbeelden beschikbaar op GitHub.