¿Cómo paralelizar el procesamiento de mensajes parcialmente ordenados?

16 Jun, 2023 Programación 0

Mi aplicación basada en Netty:

ingestiona cientos de miles de mensajes por segundo desde una sola conexión TCP.
procesa esos mensajes en un par de manejadores entrantes.
envía los resultados del procesamiento a algún lugar aguas abajo.

Actualmente, todo esto se ejecuta en un hilo, ya que está en una sola conexión TCP. Me gustaría saber cómo puedo paralelizar el punto 2. La dificultad es que los mensajes no pueden procesarse en paralelo sin cuidado, porque hay un orden parcial de los mensajes. Se puede pensar en esto como si hubiera una función clave(mensaje), y todos los mensajes para los cuales esta función devuelve el mismo resultado deben procesarse secuencialmente, pero si los resultados son diferentes, pueden ejecutarse en paralelo. Entonces, estoy pensando en tener una asignación de hilo como hash(clave(mensaje)) % recuentoHilos.

Imagínese esta canalización:

java pipeline.addLast(deframer); pipeline.addLast(new IdleStateHandler(...)); pipeline.addLast(decoder); pipeline.addLast(bizLogicHandler1); pipeline.addLast(bizLogicHandler2);

En el decodificador, puedo calcular el resultado de clave(mensaje), así que me gustaría paralelizar todo lo que está aguas abajo del decodificador. Está documentado que para utilizar varios hilos puedo hacer lo siguiente:

grupo estático final EventExecutorGroup = new DefaultEventExecutorGroup(16); ... pipeline.addLast(group, "bizLogicHandler1", bizLogicHandler1); pipeline.addLast("bizLogicHandler2", bizLogicHandler2);

lo que supongo que significa que bizLogicHandler1 y todo lo que está por debajo de él (en el ejemplo anterior, eso sería bizLogicHandler2) podrán ejecutarse en paralelo? (¿O tendría que especificar group para bizLogicHandler2 también?)

Sin embargo, lo anterior seguirá funcionando completamente en serie, como explica la documentación, ofreciendo UnorderedThreadPoolEventExecutor como alternativa para maximizar el paralelismo, a costa de eliminar completamente la ordenación, lo que no funciona en mi caso.

Al mirar las interfaces EventExecutorGroup y EventExecutor, no veo cómo sería posible transmitir qué mensajes se pueden procesar en paralelo y cuáles deben procesarse secuencialmente.

¿Algún idea?

davy.ai

June 16, 2023 at 3:10 am

Como experto en IT especializado en Netty, aquí están mis sugerencias para paralelizar el procesamiento de mensajes en tu aplicación mientras se mantiene el orden parcial de los mismos:

Puedes crear un pool de hilos personalizado que tenga en cuenta el orden parcial de los mensajes. Este pool de hilos constará de múltiples hilos, y cada hilo será responsable de procesar mensajes cuyas claves mapean al mismo valor hash. El número de hilos en el pool puede especificarse en función del número de valores hash únicos devueltos por la función de clave.
En tu decodificador, después de calcular la clave para cada mensaje, puedes enviar el mensaje al pool de hilos para su procesamiento. Puedes utilizar el método submit() de la interfaz java.util.concurrent.ExecutorService para enviar el mensaje al pool de hilos. Este método devuelve un objeto Future, que puedes utilizar para recuperar el resultado de la computación.
Una vez que se completa el procesamiento de un mensaje, puedes enviar el resultado aguas abajo utilizando un canal Netty. Puedes utilizar el método write() de la clase io.netty.channel.ChannelHandlerContext para escribir el resultado en el canal. Netty se encargará del envío real de los datos a través de la red.
En tu pipeline, debes eliminar los manejadores bizLogicHandler1 y bizLogicHandler2, ya que serán reemplazados por el pool de hilos personalizado. Luego puedes agregar un nuevo manejador que recupera los resultados de las computaciones de los objetos Future y los envía aguas abajo utilizando el canal Netty.

Con estos cambios, tu aplicación podrá procesar mensajes en paralelo mientras se mantiene el orden parcial de los mismos.

Answer

davy.ai

June 16, 2023 at 3:10 am
Como experto en IT especializado en Netty, aquí están mis sugerencias para paralelizar el procesamiento de mensajes en tu aplicación mientras se mantiene el orden parcial de los mismos:
1. Puedes crear un pool de hilos personalizado que tenga en cuenta el orden parcial de los mensajes. Este pool de hilos constará de múltiples hilos, y cada hilo será responsable de procesar mensajes cuyas claves mapean al mismo valor hash. El número de hilos en el pool puede especificarse en función del número de valores hash únicos devueltos por la función de clave.
2. En tu decodificador, después de calcular la clave para cada mensaje, puedes enviar el mensaje al pool de hilos para su procesamiento. Puedes utilizar el método submit() de la interfaz java.util.concurrent.ExecutorService para enviar el mensaje al pool de hilos. Este método devuelve un objeto Future, que puedes utilizar para recuperar el resultado de la computación.
3. Una vez que se completa el procesamiento de un mensaje, puedes enviar el resultado aguas abajo utilizando un canal Netty. Puedes utilizar el método write() de la clase io.netty.channel.ChannelHandlerContext para escribir el resultado en el canal. Netty se encargará del envío real de los datos a través de la red.
4. En tu pipeline, debes eliminar los manejadores bizLogicHandler1 y bizLogicHandler2, ya que serán reemplazados por el pool de hilos personalizado. Luego puedes agregar un nuevo manejador que recupera los resultados de las computaciones de los objetos Future y los envía aguas abajo utilizando el canal Netty.
Con estos cambios, tu aplicación podrá procesar mensajes en paralelo mientras se mantiene el orden parcial de los mismos.

Comments are closed.

es.davy.ai

¿Tienes una pregunta?

¿Cómo paralelizar el procesamiento de mensajes parcialmente ordenados?

Answer

davy.ai