java - que - Vert.x Event loop-¿Cómo es esto asíncrono?

vertx webclient get

Question

Estoy jugando con Vert.x y soy bastante nuevo en los servidores basados ​​en el bucle de eventos en lugar del modelo de conexión / subproceso.

public void start(Future<Void> fut) {
    vertx
        .createHttpServer()
        .requestHandler(r -> {
            LocalDateTime start = LocalDateTime.now();
            System.out.println("Request received - "+start.format(DateTimeFormatter.ISO_DATE_TIME));
            final MyModel model = new MyModel();
            try {

                for(int i=0;i<10000000;i++){
                    //some simple operation
                }

                model.data = start.format(DateTimeFormatter.ISO_DATE_TIME) +" - "+LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME);

            } catch (Exception e1) {
                // TODO Auto-generated catch block
                e1.printStackTrace();
            }

          r.response().end(
                  new Gson().toJson(model)
                 );
        })
        .listen(4568, result -> {
          if (result.succeeded()) {
            fut.complete();
          } else {
            fut.fail(result.cause());
          }
        });
    System.out.println("Server started ..");
  }

• Solo estoy tratando de simular un controlador de solicitud de larga ejecución para comprender cómo funciona este modelo.
• Lo que he observado es que el llamado bucle de eventos está bloqueado hasta que se complete mi primera solicitud. Independientemente del poco tiempo que tome, la solicitud posterior no se ejecutará hasta que se complete la anterior.
• Obviamente me falta una pieza aquí y esa es la pregunta que tengo aquí.

Editado en base a las respuestas hasta el momento:

1. ¿No se acepta que todas las solicitudes son asíncronas? Si solo se puede aceptar una nueva conexión cuando se borra la anterior, ¿cómo es asíncrono?
   • Suponga que una solicitud típica toma entre 100 ms y 1 segundo (según el tipo y la naturaleza de la solicitud). Por lo tanto, significa que el bucle de eventos no puede aceptar una nueva conexión hasta que finalice la solicitud anterior (incluso si termina en un segundo). Y si yo, como programador, tengo que pensar en todo esto y enviar esos manejadores de solicitudes a un subproceso de trabajo, ¿en qué se diferencia de un modelo de hilo / conexión?
   • ¿Estoy tratando de entender cómo es mejor este modelo que los modelos tradicionales de servidor de hilos / conn? ¿Supongamos que no hay ninguna operación de E / S o que todas las operaciones de E / S se manejan de forma asíncrona? ¿Cómo soluciona incluso el problema de c10k, cuando no puede iniciar todas las solicitudes concurrentes de manera paralela y tiene que esperar hasta que termine la anterior?

2. Incluso si decido enviar todas estas operaciones a un subproceso de trabajo (agrupado), entonces vuelvo al mismo problema, ¿no es así? ¿Cambio de contexto entre hilos? Edita y encabeza esta pregunta para obtener una recompensa.

   • No entiendo completamente cómo este modelo es asincrónico.
   • Vert.x tiene un cliente JDBC asíncrono (Asyncronous es la palabra clave) que intenté adaptar con RXJava.
   • Aquí hay un ejemplo de código (porciones relevantes)

   server.requestStream (). toObservable (). subscribe (req -> {

           LocalDateTime start = LocalDateTime.now();
           System.out.println("Request for " + req.absoluteURI() +" received - " +start.format(DateTimeFormatter.ISO_DATE_TIME));
           jdbc.getConnectionObservable().subscribe(
                   conn -> {
   
                       // Now chain some statements using flatmap composition
                       Observable<ResultSet> resa = conn.queryObservable("SELECT * FROM CALL_OPTION WHERE UNDERLYING=''NIFTY''");
                       // Subscribe to the final result
                       resa.subscribe(resultSet -> {
   
                           req.response().end(resultSet.getRows().toString());
                           System.out.println("Request for " + req.absoluteURI() +" Ended - " +LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
                       }, err -> {
                           System.out.println("Database problem");
                           err.printStackTrace();
                       });
                   },
   
                   // Could not connect
                   err -> {
                       err.printStackTrace();
                   }
                   );
   
   });
   server.listen(4568);

   • La consulta de selección allí toma 3 segundos aproximadamente para devolver el volcado completo de la tabla.
   • Cuando disparo solicitudes concurrentes (traté con solo 2), veo que la segunda solicitud espera completamente a que se complete la primera.
   • Si la selección de JDBC es asíncrona, ¿no es una expectativa razonable que el marco maneje la segunda conexión mientras espera que la consulta de selección devuelva algo?

Answer 1

¿Cómo es esto asíncrono? La respuesta está en tu propia pregunta.

Lo que he observado es que el llamado bucle de eventos está bloqueado hasta que se complete mi primera solicitud. Independientemente del poco tiempo que tome, la solicitud posterior no se ejecutará hasta que se complete la anterior.

La idea es que, en lugar de tener un nuevo servidor para cada solicitud HTTP, se use el mismo subproceso que ha bloqueado en su tarea de larga duración.

El objetivo del bucle de eventos es guardar el tiempo involucrado en el cambio de contexto de un hilo a otro y utilizar el tiempo de CPU ideal cuando una tarea utiliza actividades de IO / red. Si durante la gestión de su solicitud se debió a otra operación de E / S de red, por ejemplo: recuperar datos de una instancia remota de MongoDB durante ese tiempo, su subproceso no se bloqueará y, en su lugar, otra solicitud será atendida por el mismo subproceso, que es el caso de uso ideal de modelo de bucle de eventos (considerando que tiene solicitudes concurrentes que llegan a su servidor).

Si tiene tareas de ejecución prolongada que no involucran la operación de Red / IO, debería considerar usar un grupo de subprocesos, si bloquea el subproceso del bucle de evento principal, otras solicitudes se retrasarán. es decir, para las tareas de ejecución prolongada, puede pagar el precio del cambio de contexto para que el servidor responda.

EDITAR : la forma en que un servidor puede manejar las solicitudes puede variar:

1) Generar un nuevo hilo para cada solicitud entrante (En este modelo, el cambio de contexto sería alto y hay un costo adicional de generar un nuevo hilo cada vez)

2) Use un grupo de subprocesos para servir la solicitud (el mismo conjunto de subprocesos se usaría para atender solicitudes y las solicitudes adicionales se ponen en cola)

3) Use un bucle de eventos (un solo hilo para todas las solicitudes. Conmutación de contexto insignificante. Debido a que habría algunos hilos en ejecución, por ejemplo: para poner en cola las solicitudes entrantes)

En primer lugar, el cambio de contexto no es malo, se requiere para que el servidor de aplicaciones responda, pero, el cambio de contexto puede ser un problema si el número de solicitudes concurrentes es demasiado alto (aproximadamente más de 10k). Si desea comprender con más detalle, le recomiendo que lea el artículo de C10K

Suponga que una solicitud típica toma entre 100 ms y 1 segundo (según el tipo y la naturaleza de la solicitud). Por lo tanto, significa que el bucle de eventos no puede aceptar una nueva conexión hasta que finalice la solicitud anterior (incluso si termina en un segundo).

Si necesita responder a un gran número de solicitudes simultáneas (más de 10k) consideraría más de 500 ms como una operación de ejecución más larga. En segundo lugar, como dije, hay algunos subprocesos / cambio de contexto involucrados, por ejemplo: para poner en cola las solicitudes entrantes, pero, el cambio de contexto entre los subprocesos se reduciría en gran medida, ya que habría muy pocos subprocesos a la vez. En tercer lugar, si hay una operación de red / IO involucrada en la resolución de la primera solicitud, la segunda solicitud tendrá una oportunidad de resolverse antes de que se resuelva la primera, aquí es donde este modelo funciona bien.

Y si yo, como programador, tengo que pensar en todo esto y enviar esos manejadores de solicitudes a un subproceso de trabajo, ¿en qué se diferencia de un modelo de hilo / conexión?

Vertx está tratando de ofrecerle lo mejor de los subprocesos y el bucle de eventos, por lo que, como programador, puede hacer una llamada sobre cómo hacer que su aplicación sea eficiente tanto en el escenario como en la operación de larga ejecución con y sin operación de red / IO.

¿Estoy tratando de entender cómo es mejor este modelo que los modelos tradicionales de servidor de hilos / conn? ¿Supongamos que no hay ninguna operación de E / S o que todas las operaciones de E / S se manejan de forma asíncrona? ¿Cómo soluciona incluso el problema de c10k, cuando no puede iniciar todas las solicitudes concurrentes de manera paralela y tiene que esperar hasta que termine la anterior?

La explicación anterior debe responder a esto.

Incluso si decido enviar todas estas operaciones a un subproceso de trabajo (agrupado), entonces vuelvo al mismo problema, ¿no es así? ¿Cambio de contexto entre hilos?

Como dije, ambos tienen ventajas y desventajas, y vertx te ofrece el modelo y, dependiendo de tu caso de uso, debes elegir lo que es ideal para tu escenario.

Answer 2

El bucle de eventos Vert.x es, de hecho, un bucle de eventos clásico que existe en muchas plataformas. Y, por supuesto, la mayoría de las explicaciones y documentos se pueden encontrar para Node.js, ya que es el marco más popular basado en este patrón de arquitectura. Eche un vistazo a una explanation más o menos buena de la mecánica en el bucle de eventos Node.js. El tutorial de Vert.x tiene una buena explicación entre "No nos llames, te llamaremos" y "Verticles" también.

Editar para sus actualizaciones:

En primer lugar, cuando se trabaja con un bucle de eventos, el hilo principal debería funcionar muy rápidamente para todas las solicitudes. No deberías hacer ningún trabajo largo en este bucle. Y, por supuesto, no debe esperar la respuesta de su llamada a la base de datos. - Programar una llamada de forma asíncrona - Asignar una devolución de llamada (manejador) al resultado - La devolución de llamada se ejecutará en un subproceso de trabajo, un subproceso de bucle de evento no. Esta devolución de llamada, por ejemplo, devolverá la respuesta al socket. Por lo tanto, sus operaciones en el bucle de eventos solo deben programar todas las operaciones asíncronas con devoluciones de llamada y pasar a la siguiente solicitud sin esperar ningún resultado.

Suponga que una solicitud típica toma entre 100 ms y 1 segundo (según el tipo y la naturaleza de la solicitud).

En ese caso, su solicitud tiene partes costosas de cómputo o acceso a IO: su código en el bucle de eventos no debe esperar el resultado de estas operaciones.

¿Estoy tratando de entender cómo es mejor este modelo que los modelos tradicionales de servidor de hilos / conn? ¿Supongamos que no hay ninguna operación de E / S o que todas las operaciones de E / S se manejan de forma asíncrona?

Cuando tenga demasiadas solicitudes simultáneas y un modelo de programación tradicional, creará un hilo por cada solicitud. ¿Qué hará este hilo? En su mayoría serán esperables para las operaciones de E / S (por ejemplo, resultado de la base de datos). Es un desperdicio de recursos. En nuestro modelo de bucle de eventos, tiene un hilo principal que programa las operaciones y preasigna la cantidad de hilos de trabajo para tareas largas. + Ninguno de estos trabajadores realmente espera una respuesta, solo puede ejecutar otro código mientras espera el resultado de IO (puede implementarse como devoluciones de llamada o estado de verificación periódica de trabajos de IO actualmente en curso). Le recomendaría ir a través de Java NIO y Java NIO 2 para comprender cómo este IO asíncrono puede implementarse realmente dentro del marco. Los hilos verdes también son conceptos muy relacionados, eso sería bueno entenderlos. Los subprocesos verdes y las rutinas son un tipo de bucle de eventos sombreados, que intentan lograr lo mismo, menos subprocesos, porque podemos reutilizar el subproceso del sistema mientras que los subprocesos verdes esperan algo.

¿Cómo soluciona incluso el problema de c10k, cuando no puede iniciar todas las solicitudes simultáneas en paralelo y tiene que esperar hasta que termine la anterior?

Seguro que no esperamos en el hilo principal para enviar respuesta de solicitud previa. Obtener solicitud, programar la ejecución de tareas largas / IO, siguiente solicitud.

Incluso si decido enviar todas estas operaciones a un subproceso de trabajo (agrupado), entonces vuelvo al mismo problema, ¿no es así? ¿Cambio de contexto entre hilos?

Si haces todo bien, no. Aún más, obtendrá una buena localización de datos y predicción de flujo de ejecución. Un núcleo de CPU ejecutará su ciclo de eventos cortos y programará el trabajo asíncrono sin cambio de contexto y nada más. Otros núcleos hacen llamada a la base de datos y devuelven la respuesta y solo esto. La conmutación entre devoluciones de llamada o la comprobación de diferentes canales para el estado de IO en realidad no requiere el cambio de contexto de cualquier subproceso del sistema; en realidad, funciona en un subproceso de trabajo. Por lo tanto, tenemos un subproceso de trabajo por núcleo y este subproceso del sistema espera / verifica la disponibilidad de resultados de múltiples conexiones a la base de datos, por ejemplo. Revise el concepto de Java NIO para comprender cómo puede funcionar de esta manera. (Ejemplo clásico para NIO: servidor proxy que puede aceptar muchas conexiones paralelas (miles), solicitudes de proxy a otros servidores remotos, escuchar respuestas y enviar respuestas a los clientes y todo esto utilizando uno o dos subprocesos)

Acerca de su código, hice un project muestra para que usted demuestre que todo funciona como se esperaba:

public class MyFirstVerticle extends AbstractVerticle {

    @Override
    public void start(Future<Void> fut) {
        JDBCClient client = JDBCClient.createShared(vertx, new JsonObject()
                .put("url", "jdbc:hsqldb:mem:test?shutdown=true")
                .put("driver_class", "org.hsqldb.jdbcDriver")
                .put("max_pool_size", 30));


        client.getConnection(conn -> {
            if (conn.failed()) {throw new RuntimeException(conn.cause());}
            final SQLConnection connection = conn.result();

            // create a table
            connection.execute("create table test(id int primary key, name varchar(255))", create -> {
                if (create.failed()) {throw new RuntimeException(create.cause());}
            });
        });

        vertx
            .createHttpServer()
            .requestHandler(r -> {
                int requestId = new Random().nextInt();
                System.out.println("Request " + requestId + " received");
                    client.getConnection(conn -> {
                         if (conn.failed()) {throw new RuntimeException(conn.cause());}

                         final SQLConnection connection = conn.result();

                         connection.execute("insert into test values (''" + requestId + "'', ''World'')", insert -> {
                             // query some data with arguments
                             connection
                                 .queryWithParams("select * from test where id = ?", new JsonArray().add(requestId), rs -> {
                                     connection.close(done -> {if (done.failed()) {throw new RuntimeException(done.cause());}});
                                     System.out.println("Result " + requestId + " returned");
                                     r.response().end("Hello");
                                 });
                         });
                     });
            })
            .listen(8080, result -> {
                if (result.succeeded()) {
                    fut.complete();
                } else {
                    fut.fail(result.cause());
                }
            });
    }
}

@RunWith(VertxUnitRunner.class)
public class MyFirstVerticleTest {

  private Vertx vertx;

  @Before
  public void setUp(TestContext context) {
    vertx = Vertx.vertx();
    vertx.deployVerticle(MyFirstVerticle.class.getName(),
        context.asyncAssertSuccess());
  }

  @After
  public void tearDown(TestContext context) {
    vertx.close(context.asyncAssertSuccess());
  }

  @Test
  public void testMyApplication(TestContext context) {
      for (int i = 0; i < 10; i++) {
          final Async async = context.async();
          vertx.createHttpClient().getNow(8080, "localhost", "/",
                            response -> response.handler(body -> {
                                context.assertTrue(body.toString().contains("Hello"));
                                async.complete();
                            })
        );
    }
  }
}

Salida:

Request 1412761034 received
Request -1781489277 received
Request 1008255692 received
Request -853002509 received
Request -919489429 received
Request 1902219940 received
Request -2141153291 received
Request 1144684415 received
Request -1409053630 received
Request -546435082 received
Result 1412761034 returned
Result -1781489277 returned
Result 1008255692 returned
Result -853002509 returned
Result -919489429 returned
Result 1902219940 returned
Result -2141153291 returned
Result 1144684415 returned
Result -1409053630 returned
Result -546435082 returned

Por lo tanto, aceptamos solicitud: programamos la solicitud a la base de datos, pasamos a la siguiente solicitud, los consumimos todos y enviamos respuesta para cada solicitud solo cuando todo se hace con la base de datos.

Acerca del ejemplo de código, veo dos posibles problemas: primero, parece que no close() conexión, lo cual es importante para devolverlo al grupo. Segundo, ¿cómo se configura tu pool? Si solo hay una conexión libre, estas solicitudes se serializarán esperando esta conexión.

Le recomiendo que agregue un poco de impresión de la marca de tiempo para que ambas solicitudes encuentren el lugar donde se serializa. Usted pone algo que hace llamadas en el bloqueo del bucle de eventos. O ... compruebe que envía solicitudes en paralelo en su prueba. No sigue después de recibir respuesta después de anterior.

Answer 3

En este tipo de motores de procesamiento, se supone que debe convertir las tareas de ejecución prolongada en operaciones ejecutadas de forma asíncrona y esta es una metodología para hacer esto, de modo que el hilo crítico pueda completarse lo más rápido posible y volver a realizar otra tarea. es decir, cualquier operación de IO se pasa al marco para devolverle la llamada cuando finaliza el IO.

El marco es asíncrono en el sentido de que le permite producir y ejecutar estas tareas asíncronas, pero no cambia su código de ser síncrono a asíncrono.