Implementing a Ring Buffer in Java

1. Áttekintés

Ezzel a bemutatóval megtanuljuk, hogyan kell egy Ring Buffer-t Java-ban implementálni.

Ring Buffer

A Ring Buffer (vagy Circular Buffer) egy körülhatárolt kör alakú adatstruktúra, amelyet két vagy több szál közötti adatok pufferelésére használnak. Mivel folyamatosan írunk egy gyűrűs pufferbe, az a végére érve körbefordul.

2.1. Hogyan működik

A gyűrűpuffer egy fix méretű tömb segítségével valósul meg, amely a határainál körbetekeredik.

A tömbön kívül három dolgot tart számon:

• a következő szabad helyet a pufferben egy elem beillesztéséhez,
• a következő olvasatlan elemet a pufferben,
• és a tömb végét – azt a pontot, ahol a puffer körbetekeredik a tömb elejére

Az, hogy a gyűrűs puffer hogyan kezeli ezeket a követelményeket, az implementációtól függően változik. A Wikipédia bejegyzése a témában például egy négymutatós módszert mutat be.

A megközelítést a Disruptor szekvenciákat használó gyűrűpuffer implementációjából kölcsönözzük.

Az első dolog, amit tudnunk kell, az a kapacitás – a puffer rögzített maximális mérete. Ezután két monoton növekvő szekvenciát fogunk használni:

• Sorozat írása: -1-nél kezdődik, 1-gyel növekszik, amikor elemet illesztünk be
• Sorozat olvasása: 0-nál kezdődik, 1-gyel növekszik, amikor elemet fogyasztunk

A szekvenciát egy mod művelettel leképezhetjük a tömb egy indexére:

arrayIndex = sequence % capacity

A mod művelet a szekvenciát a határok köré tekeri, hogy a pufferben levezethessünk egy helyet:

Lássuk, hogyan illesztünk be egy elemet:

buffer = element

Elem beillesztése előtt előre inkrementáljuk a szekvenciát.

Az elem fogyasztásához utólagos inkrementálást végzünk:

element = buffer

Ebben az esetben utólagos inkrementálást végzünk a szekvencián. Egy elem elfogyasztása nem távolítja el azt a pufferből – csak addig marad a tömbben, amíg felül nem írjuk.

2.2. Az elemet a pufferben tartjuk. Üres és teli pufferek

Amint körbetekerjük a tömböt, elkezdjük felülírni a pufferben lévő adatokat. Ha a puffer tele van, választhatunk, hogy vagy felülírjuk a legrégebbi adatokat, függetlenül attól, hogy az olvasó elfogyasztotta-e azokat, vagy megakadályozzuk a még be nem olvasott adatok felülírását.

Ha az olvasó megengedheti magának, hogy a köztes vagy régi értékek kimaradjanak (például egy tőzsdei árfolyamketyegő), akkor az adatokat felülírhatjuk anélkül, hogy megvárnánk, amíg elfogynak. Másrészt, ha az olvasónak az összes értéket el kell fogyasztania (mint például az e-kereskedelmi tranzakciók esetében), akkor várnunk kell (block/busy-wait), amíg a pufferben van szabad slot.

A puffer tele van, ha a puffer mérete megegyezik a kapacitásával, ahol a mérete megegyezik az olvasatlan elemek számával:

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

Ha az írási sorozat elmarad az olvasási sorozattól, a puffer üres:

isEmpty = writeSequence < readSequence

A puffer null értéket ad vissza, ha üres.

2.2. Az írási sorozatot az olvasási sorozattól elmaradva a puffer üres. Előnyök és hátrányok

A gyűrűs puffer egy hatékony FIFO puffer. Egy fix méretű tömböt használ, amelyet előre ki lehet osztani, és hatékony memória-hozzáférési mintát tesz lehetővé. Minden pufferművelet állandó idejű O(1), beleértve egy elem elfogyasztását is, mivel nem igényel elemeltolást.

A másik oldalon a gyűrűpuffer megfelelő méretének meghatározása kritikus. Például az írási műveletek hosszú ideig blokkolhatnak, ha a puffer alulméretezett, és az olvasás lassú. Használhatunk dinamikus méretezést, de ehhez az adatok mozgatására lenne szükség, és a fentebb tárgyalt előnyök nagy részéről lemaradnánk.

Implementálás Javában

Most, hogy megértettük, hogyan működik a gyűrűpuffer, folytassuk a megvalósítását Javában.

3.1. A gyűrűpuffer megvalósítása. Inicializálás

Először definiáljunk egy konstruktort, amely inicializálja a puffert egy előre meghatározott kapacitással:

public CircularBuffer(int capacity) { this.capacity = (capacity < 1) ? DEFAULT_CAPACITY : capacity; this.data = (E) new Object; this.readSequence = 0; this.writeSequence = -1;}

Ez egy üres puffert hoz létre, és inicializálja a szekvencia mezőit az előző szakaszban tárgyaltak szerint.

3.2. Inicializálás. Ajánlat

A következőkben implementáljuk az ajánlat műveletet, amely egy elemet illeszt be a pufferbe a következő szabad helyre, és siker esetén true-t ad vissza. Fals értéket ad vissza, ha a pufferben nem találunk üres slotot, vagyis nem tudjuk felülírni az olvasatlan értékeket.

Implumentáljuk az offer metódust Java nyelven:

public boolean offer(E element) { boolean isFull = (writeSequence - readSequence) + 1 == capacity; if (!isFull) { int nextWriteSeq = writeSequence + 1; data = element; writeSequence++; return true; } return false;}

Az írási sorrendet inkrementáljuk, és kiszámítjuk az indexet a tömbben a következő szabad slothoz. Ezután írjuk az adatokat a pufferbe, és tároljuk a frissített írási sorrendet.

Kipróbáljuk:

@Testpublic void givenCircularBuffer_whenAnElementIsEnqueued_thenSizeIsOne() { CircularBuffer buffer = new CircularBuffer<>(defaultCapacity); assertTrue(buffer.offer("Square")); assertEquals(1, buffer.size());}

3.3. Az írási sorrendet a pufferbe írjuk. Poll

Végül implementáljuk a poll műveletet, amely lekérdezi és eltávolítja a következő olvasatlan elemet. A poll művelet nem távolítja el az elemet, hanem növeli az olvasási sorrendet.

Elvégre implementáljuk:

public E poll() { boolean isEmpty = writeSequence < readSequence; if (!isEmpty) { E nextValue = data; readSequence++; return nextValue; } return null;}

Itt a tömbben lévő index kiszámításával olvassuk az adatokat az aktuális olvasási sorrendnél. Ezután inkrementáljuk a sorrendet, és visszaadjuk az értéket, ha a puffer nem üres.

Teszteljük ki:

@Testpublic void givenCircularBuffer_whenAnElementIsDequeued_thenElementMatchesEnqueuedElement() { CircularBuffer buffer = new CircularBuffer<>(defaultCapacity); buffer.offer("Triangle"); String shape = buffer.poll(); assertEquals("Triangle", shape);}

Producer-Fogyasztó probléma

A gyűrűs puffer használatáról beszéltünk két vagy több szál közötti adatcserére, ami a Producer-Fogyasztó probléma nevű szinkronizációs probléma egyik példája. Javában a producer-fogyasztó problémát többféleképpen is megoldhatjuk szemaforok, korlátos várólisták, gyűrűs pufferek stb. segítségével.

Impalmentáljunk egy gyűrűs pufferre épülő megoldást.

4.1. Volatile Sequence Fields

A gyűrűs puffer implementációnk nem szálbiztos. Tegyük szálbiztossá az egyszerű egytermelős és egyfogyasztós esetre.

A termelő adatokat ír a pufferbe, és növeli a writeSequence-t, míg a fogyasztó csak olvas a pufferből, és növeli a readSequence-t. A pufferben csak a readSequence-t növeli. A háttértömb tehát versengésmentes, és megúszhatjuk szinkronizálás nélkül.

Mégis biztosítanunk kell, hogy a fogyasztó lássa a writeSequence mező legutóbbi értékét (láthatóság), és hogy a writeSequence ne frissüljön, mielőtt az adat ténylegesen elérhető lenne a pufferben (sorrendiség).

A gyűrűs puffert ebben az esetben úgy tehetjük konkurenssé és lock-mentessé, hogy a szekvencia mezők volatilevé válnak:

private volatile int writeSequence = -1, readSequence = 0;

A kínálati metódusban a volatile writeSequence mezőbe való írás garantálja, hogy a pufferbe való írás a szekvencia frissítése előtt történik. Ugyanakkor az illékony láthatósági garancia biztosítja, hogy a fogyasztó mindig a writeSequence legfrissebb értékét lássa.

4.2. Az illékony láthatóság garantálása. Producer

Implementáljunk egy egyszerű producer Runnable-t, amely a gyűrűs pufferbe ír:

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

A producer szál egy ciklusban várna egy üres résre (busy-waiting).

4.3. Fogyasztó

Egy fogyasztói hívhatót valósítunk meg, amely a pufferből olvas:

public T call() { T items = (T) new Object; for (int i = 0; i < items.length;) { T item = buffer.poll(); if (item != null) { items = item; System.out.println("Consumed: " + item); } } return items;}

A fogyasztói szál nyomtatás nélkül folytatja, ha null értéket kap a pufferből.

Írjuk meg a meghajtó kódunkat:

executorService.submit(new Thread(new Producer<String>(buffer)));executorService.submit(new Thread(new Consumer<String>(buffer)));

A termelő-fogyasztó programunk végrehajtása az alábbi kimenetet eredményezi:

Produced: CircleProduced: Triangle Consumed: CircleProduced: Rectangle Consumed: Triangle Consumed: RectangleProduced: SquareProduced: Rhombus Consumed: SquareProduced: Trapezoid Consumed: Rhombus Consumed: TrapezoidProduced: PentagonProduced: PentagramProduced: Hexagon Consumed: Pentagon Consumed: PentagramProduced: Hexagram Consumed: Hexagon Consumed: Hexagram

Következtetés

Ezzel a bemutatóval megtanultuk, hogyan kell megvalósítani egy gyűrűs puffert, és feltártuk, hogyan használható a termelő-fogyasztó probléma megoldására.