streamalism.org

1. Overzicht

In deze tutorial zullen we kijken naar java.util.concurrent.Exchanger. Dit werkt als een gemeenschappelijk punt voor twee threads in Java om onderling objecten uit te wisselen.

2. Inleiding tot wisselaar

De Uitwisselaar class in Java kan worden gebruikt om objecten tussen twee threads van het type te delen T. De klasse biedt slechts één overbelaste methode uitwisseling (T t).

Wanneer aangeroepen uitwisseling wacht op de andere thread in het paar om het ook te bellen. Op dit punt merkt de tweede thread dat de eerste thread wacht met zijn object. De draad wisselt de voorwerpen die ze vasthouden en signaleert de uitwisseling, en nu kunnen ze terugkeren.

Laten we een voorbeeld bekijken om de berichtuitwisseling tussen twee threads te begrijpen met Uitwisselaar:

@Test openbare ongeldig gegevenThreads_whenMessageExchanged_thenCorrect () {Wisselaarwisselaar = nieuwe wisselaar (); Runnable taskA = () -> {probeer {String message = exchanger.exchange ("van A"); assertEquals ("van B", bericht); } catch (InterruptedException e) {Thread.currentThread.interrupt (); gooi nieuwe RuntimeException (e); }}; Runnable taskB = () -> {probeer {String message = exchanger.exchange ("van B"); assertEquals ("van A", bericht); } catch (InterruptedException e) {Thread.currentThread.interrupt (); gooi nieuwe RuntimeException (e); }}; CompletableFuture.allOf (runAsync (taskA), runAsync (taskB)). Join (); }

Hier hebben we de twee threads die berichten tussen elkaar uitwisselen met behulp van de gemeenschappelijke wisselaar. Laten we eens kijken naar een voorbeeld waarin we een object uit de hoofdthread uitwisselen met een nieuwe thread:

@Test openbare ongeldig gegevenThread_WhenExchangedMessage_thenCorrect () gooit InterruptedException {Exchanger exchanger = new Exchanger (); Runnable runner = () -> {probeer {String message = exchanger.exchange ("from runner"); assertEquals ("to runner", bericht); } catch (InterruptedException e) {Thread.currentThread.interrupt (); gooi nieuwe RuntimeException (e); }}; CompletableFuture resultaat = CompletableFuture.runAsync (runner); String msg = exchanger.exchange ("to runner"); assertEquals ("from runner", msg); resultaat.join (); }

Merk op dat we het loper draad eerst en later bellen uitwisseling() in de hoofddraad.

Houd er ook rekening mee dat de aanroep van de eerste thread een time-out kan veroorzaken als de tweede thread het uitwisselingspunt niet op tijd bereikt. Hoe lang de eerste thread moet wachten, kan worden gecontroleerd met behulp van de overbelaste exchange (T t, lange time-out, TimeUnit timeUnit).

3. Geen GC-gegevensuitwisseling

Uitwisselaar kan worden gebruikt om pijplijnpatronen te creëren waarbij gegevens van de ene thread naar de andere worden doorgegeven. In deze sectie maken we een eenvoudige stapel threads die continu gegevens als een pijplijn tussen elkaar doorgeven.

@Test openbare leegte gegevenData_whenPassedThrough_thenCorrect () gooit InterruptedException {Exchanger readerExchanger = nieuwe wisselaar (); Uitwisselaar writerExchanger = nieuwe wisselaar (); Runnable reader = () -> {Wachtrij readerBuffer = nieuwe ConcurrentLinkedQueue (); while (true) {readerBuffer.add (UUID.randomUUID (). toString ()); if (readerBuffer.size ()> = BUFFER_SIZE) {readerBuffer = readerExchanger.exchange (readerBuffer); }}}; Runnable processor = () -> {Queue processorBuffer = nieuwe ConcurrentLinkedQueue (); Wachtrij writerBuffer = nieuwe ConcurrentLinkedQueue (); processorBuffer = readerExchanger.exchange (processorBuffer); while (true) {writerBuffer.add (processorBuffer.poll ()); if (processorBuffer.isEmpty ()) {processorBuffer = readerExchanger.exchange (processorBuffer); writerBuffer = writerExchanger.exchange (writerBuffer); }}}; Runnable writer = () -> {Wachtrij writerBuffer = nieuwe ConcurrentLinkedQueue (); writerBuffer = writerExchanger.exchange (writerBuffer); while (true) {System.out.println (writerBuffer.poll ()); if (writerBuffer.isEmpty ()) {writerBuffer = writerExchanger.exchange (writerBuffer); }}}; CompletableFuture.allOf (runAsync (lezer), runAsync (processor), runAsync (schrijver)). Join (); }

Hier hebben we drie threads: lezer, processor, en auteur. Samen werken ze als een enkele pijplijn die onderling gegevens uitwisselt.

De readerExchanger wordt gedeeld tussen de lezer en de processor draad, terwijl de writerExchanger wordt gedeeld tussen de processor en de auteur draad.

Merk op dat het voorbeeld hier alleen ter demonstratie is. We moeten voorzichtig zijn bij het maken van oneindige lussen met terwijl (waar). Om de code leesbaar te houden, hebben we enkele uitzonderingen weggelaten.

Dit patroon van gegevensuitwisseling tijdens het hergebruik van de buffer zorgt voor minder garbagecollection. De uitwisselingsmethode retourneert dezelfde wachtrijinstanties en er zou dus geen GC zijn voor deze objecten. In tegenstelling tot andere blokkeerwachtrijen, maakt de wisselaar geen knooppunten of objecten om gegevens vast te houden en te delen.

Het creëren van een dergelijke pijplijn is vergelijkbaar met het Disrupter-patroon, met een belangrijk verschil: het Disrupter-patroon ondersteunt meerdere producenten en consumenten, terwijl een wisselaar kan worden gebruikt tussen een paar consumenten en producenten.

4. Conclusie

Dus we hebben geleerd wat Uitwisselaar is in Java, hoe het werkt, en we hebben gezien hoe we de Uitwisselaar klasse. We hebben ook een pijplijn gemaakt en GC-loze gegevensuitwisseling tussen threads gedemonstreerd.

Zoals altijd is de code beschikbaar op GitHub.