[vc_row][vc_column][vc_column_text]
En el siguiente artículo, presentaremos un algoritmo para un trabajar una Cola en forma concurrente que soporta un lector y múltiples hilos de escritura. Los hilos de escritura solo necesitan leer desde un campo local que sea un hilo y escribir en un atributo volatile para publicar el evento en la Cola. La escritura no necesita «comparar y cambiar» operaciones como las colas concurrentes estandard del JDK, permitiendo un algoritmo fácil y rápido. Un ejemplo de uso es un hilo escribiendo en un archivo de eventos en forma asíncrona.
[/vc_column_text][/vc_column][/vc_row][vc_row][vc_column][vc_column_text]
Escritura
La idea principal es no utilizar una simple Cola, sino muchas. Utilizamos una Cola por hilo de escritura almacenada en un campo local. A continuación, la Cola es una lista vinculada simple que utiliza un campo volatile para el siguiente elemento y un tipo final para el valor:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 | public class LinkedList implements Consumer { volatile ListElementPointer lastRead; private LinkedListElement lastWritten; // Constructor omitted @Override public void accept(T event) { // Queue stopped logic omitted LinkedListElement linkedListElement= new LinkedListElement(event); if( lastWritten == null ) { lastWritten = linkedListElement; lastRead= new ListElementPointer(lastWritten); } else { lastWritten.next = linkedListElement; lastWritten = lastWritten.next; } } } class LinkedListElement { volatile LinkedListElement next; final T event; // Constructor omitted } |
Escribir un elemento en la cola se implementa en el método accept. Cuando se escribe el primer elemento, lastWritten y lastRead se establecerán en el nuevo LinkedListElement. De lo contrario, la lista se amplia con el nuevo LinkedListElement, y lastWritten se mueve al final de la lista.
Y aquí esta la clase que almacena cada cola, llamada Consumer, en un campo thread local:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 | public class ThreadLocalConsumer implements Consumer { private final EventBus theBus; private ThreadLocal<Consumer> threadLocal = new ThreadLocal<Consumer>(); // Constructor omitted public void accept(T event) { Consumer consumer = threadLocal.get(); if( consumer == null ) { consumer = theBus.newConsumerForThreadLocalStorage(Thread.currentThread()); threadLocal.set( consumer ); } consumer.accept(event); } } |
[/vc_column_text][/vc_column][/vc_row][vc_row][vc_column][vc_column_text]
Lectura
Cuando se leen los elementos, se requiere recordar el último elemento a leer. Esto lo hacemos en el campo lastRead en LinkedList. A continuación el método de lectura utilizado para leer los elementos de una cola:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | public void prozessWithoutReadCount(EventSink eventSink) { if(list.lastRead != null ) { // First element read if( ! list.lastRead.isRead ) { eventSink.execute( list.lastRead.element.event ); list.lastRead.isRead = true; } LinkedListElement current = list.lastRead.element; while( current.next != null ) { eventSink.execute( current.event ); current = current.next; } list.lastRead.element = current; } } |
A continuación la clase ListElementPointer utilizado para almacenar el último elemento de lectura:
1 2 3 4 5 | public class ListElementPointer { LinkedListElement element; boolean isRead; // Constructor omitted } |
El atributo lastRead es inicializado por el hilo de escritura y, luego, solo se modifica mediante el hilo de lectura.
En este artículo no se muestra el código para crear nuevas listas enlazadas y leer múltiples listas enlazadas en el hilo de lectura. Puede ver el código ejemplo completo.[/vc_column_text][/vc_column][/vc_row][vc_row][vc_column][vc_column_text]
Uso
La cola es de código abierto y el código esta disponible en GitHub. Una aplicación de este algoritmo es VMLens, una herramienta que detecta errores del tipo race conditions y deadlocks en una aplicación Java. VMLens hace un seguimiento a la aplicación y escribe los eventos en forma asíncrona, utilizando esta Cola, en un archivo para posteriores análisis.
Este artículo esta basado en An Algorithm for a Concurrent Queue[/vc_column_text][/vc_column][/vc_row]
