streamalism.org

1. Oversigt

I denne vejledning ser vi på java.util.concurrent.Exchanger. Dette fungerer som et fælles punkt for to tråde i Java til at udveksle objekter mellem dem.

2. Introduktion til Exchanger

Det Veksler klasse i Java kan bruges til at dele objekter mellem to tråde af typen T. Klassen giver kun en enkelt overbelastet metode udveksling (T t).

Når påberåbt udveksling venter på, at den anden tråd i parret også kalder det. På dette tidspunkt finder den anden tråd den første tråd venter med sit objekt. Tråden udveksler de objekter, de holder, og signalerer udvekslingen, og nu kan de vende tilbage.

Lad os se på et eksempel for at forstå meddelelsesudveksling mellem to tråde med Veksler:

@Test offentlig ugyldighed givenThreads_whenMessageExchanged_thenCorrect () {Exchanger exchanger = new Exchanger (); Kørbar taskA = () -> {prøv {String message = exchanger.exchange ("fra A"); assertEquals ("fra B", besked); } fange (InterruptedException e) {Thread.currentThread.interrupt (); smid ny RuntimeException (e); }}; Kørbar taskB = () -> {prøv {String message = exchanger.exchange ("fra B"); assertEquals ("fra A", besked); } fange (InterruptedException e) {Thread.currentThread.interrupt (); smid ny RuntimeException (e); }}; CompletableFuture.allOf (runAsync (taskA), runAsync (taskB)). Join (); }

Her har vi de to tråde, der udveksler meddelelser mellem hinanden ved hjælp af den fælles veksler. Lad os se et eksempel, hvor vi udveksler et objekt fra hovedtråden med en ny tråd:

@Test offentlig ugyldighed givenThread_WhenExchangedMessage_thenCorrect () kaster InterruptedException {Exchanger exchanger = new Exchanger (); Kørbar runner = () -> {prøv {String message = exchanger.exchange ("fra runner"); assertEquals ("til løber", besked); } fange (InterruptedException e) {Thread.currentThread.interrupt (); smid ny RuntimeException (e); }}; CompletableFuture result = CompletableFuture.runAsync (runner); Streng msg = exchanger.exchange ("til løber"); assertEquals ("fra løber", msg); result.join (); }

Bemærk, at vi skal starte løber tråd først og senere opkald udveksling() i hovedtråden.

Bemærk også, at den første tråds opkald kan være timeout, hvis den anden tråd ikke når udvekslingspunktet i tide. Hvor længe den første tråd skal vente kan styres ved hjælp af overbelastet udveksling (T t, lang timeout, TimeUnit timeUnit).

3. Ingen GC-dataudveksling

Veksler kunne bruges til at skabe pipeline slags mønstre med videregivelse af data fra den ene tråd til den anden. I dette afsnit opretter vi en simpel stak af tråde, der kontinuerligt overfører data mellem hinanden som en pipeline.

@Test offentlig ugyldighed givenData_whenPassedThrough_thenCorrect () kaster InterruptedException {Exchanger readerExchanger = ny Exchanger (); Veksler writerExchanger = ny Exchanger (); Kørbar læser = () -> {KølæserBuffer = ny ConcurrentLinkedQueue (); while (true) {readerBuffer.add (UUID.randomUUID (). toString ()); hvis (readerBuffer.size ()> = BUFFER_SIZE) {readerBuffer = readerExchanger.exchange (readerBuffer); }}} Kørbar processor = () -> {Kø processorBuffer = ny ConcurrentLinkedQueue (); KøforfatterBuffer = ny ConcurrentLinkedQueue (); processorBuffer = readerExchanger.exchange (processorBuffer); while (true) {writerBuffer.add (processorBuffer.poll ()); hvis (processorBuffer.isEmpty ()) {processorBuffer = readerExchanger.exchange (processorBuffer); writerBuffer = writerExchanger.exchange (writerBuffer); }}} Kørbar forfatter = () -> {KøforfatterBuffer = ny ConcurrentLinkedQueue (); writerBuffer = writerExchanger.exchange (writerBuffer); while (true) {System.out.println (writerBuffer.poll ()); hvis (writerBuffer.isEmpty ()) {writerBuffer = writerExchanger.exchange (writerBuffer); }}} CompletableFuture.allOf (runAsync (læser), runAsync (processor), runAsync (forfatter)). Join (); }

Her har vi tre tråde: læser, processorog forfatter. Sammen arbejder de som en enkelt pipeline, der udveksler data mellem dem.

Det readerExchanger deles mellem læser og processor tråd, mens writerExchanger deles mellem processor og forfatter tråd.

Bemærk, at eksemplet her kun er til demonstration. Vi skal være forsigtige med at skabe uendelige sløjfer med mens (sandt). Også for at holde koden læselig har vi udeladt nogle undtagelseshåndtering.

Dette mønster for udveksling af data under genbrug af bufferen muliggør mindre affaldsindsamling. Udvekslingsmetoden returnerer de samme køforekomster, og der vil derfor ikke være nogen GC for disse objekter. I modsætning til enhver blokerende kø opretter veksleren ikke nogen noder eller objekter til at holde og dele data.

Oprettelse af en sådan pipeline svarer til Disrupter-mønsteret, med en nøgleforskel, Disrupter-mønsteret understøtter flere producenter og forbrugere, mens en veksler kan bruges mellem et par forbrugere og producenter.

4. Konklusion

Så vi har lært hvad Veksler er i Java, hvordan det fungerer, og vi har set, hvordan man bruger Veksler klasse. Vi oprettede også en pipeline og demonstrerede GC-mindre dataudveksling mellem tråde.

Som altid er koden tilgængelig på GitHub.