java - Vert.x事件循环-如何异步？

Question

我在玩Vert.x，并且是基于事件循环而不是线程/连接模型的服务器新手。

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 ..");
  }

我只是想模拟一个长时间运行的请求处理程序，以了解此模型的工作原理。

我观察到的是所谓的事件循环被阻塞，直到我的第一个请求完成。无论花费多少时间，后续请求都不会响应，直到前一个请求完成为止。

显然，我在这里丢失了一块，这就是我在这里的问题。

根据目前的答案进行编辑:

不接受所有被视为异步的请求吗？如果是新的
仅当清除前一个连接时才能接受连接
关闭，它如何异步？

假定一个典型的请求花费100毫秒到1秒钟的时间(根据请求的种类和性质)。所以这意味着
在上一个请求之前，事件循环无法接受新连接
完成(即使它在一秒钟内结束)。如果我是程序员
必须仔细考虑所有这些并将这些请求处理程序推送到
工作线程，那么它与线程/连接有何不同
模型？

我只是想了解与传统的线程/conn服务器模型相比，该模型如何更好？假设没有I/O操作或
所有的I/O操作都是异步处理的？怎么解决
c10k问题，当它无法并行启动所有并发请求而不得不等到前一个请求终止时？

即使我决定将所有这些操作都推送到工作线程(已池化)，但我又回到了同样的问题，不是吗？在线程之间进行上下文切换？
编辑并在此问题上方加一个悬赏

不完全了解如何将此模型声明为异步的。

Vert.x具有一个异步JDBC客户端(关键字Asyncronous)，我尝试将其与RXJava配合使用。

这是代码示例(相关部分)

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);

那里的select查询大约需要3秒钟才能返回完整的表转储。

当我启动并发请求(仅尝试2个)时，我看到第二个请求完全等待第一个请求完成。

如果JDBC select是异步的，那么在等待select查询返回任何内容时让框架处理第二个连接不是一个合理的期望吗？

Answer 1

实际上，Vert.x事件循环是许多平台上存在的经典事件循环。当然，对于Node.js可以找到大多数说明和文档，因为它是基于此体系结构模式的最流行的框架。看一下Node.js事件循环下的一种或多种好用的explanation。 Vert.x tutorial在“不要给我们打电话，我们会打电话给您”和“Verticles”之间也有很好的解释。

编辑您的更新:

首先，当您使用事件循环时，主线程应非常快速地处理所有请求。您不应在此循环中做任何长时间的工作。当然，您不应该等待对数据库调用的响应。
-异步安排通话
-为结果分配一个回调(处理程序)
-回调将在工作线程中执行，而不是在事件循环线程中执行。例如，此回调将向套接字返回响应。
因此，事件循环中的操作应该只计划所有带有回调的异步操作，然后转到下一个请求，而无需等待任何结果。

Assume a typical request takes anywhere between 100 ms to 1 sec (based on the kind and nature of the request).

在这种情况下，您的请求将包含一些计算量大的部分或对IO的访问-事件循环中的代码不应等待这些操作的结果。

I'm just trying to understand how is this model better from a traditional thread/conn server models? Assume there is no I/O op or all the I/O op are handled asynchronously?

当并发请求太多并且使用传统的编程模型时，将为每个请求创建线程。这个线程会做什么？他们将主要等待IO操作(例如，数据库结果)。这是浪费资源。在我们的事件循环模型中，您有一个主线程来计划操作和为长任务预先分配的工作线程数量。 +这些工作人员实际上都没有等待响应，他们只能在等待IO结果的同时执行另一个代码(可以将其实现为当前正在进行的IO作业的回调或定期检查状态)。我建议您通过 Java NIO和Java NIO 2来了解如何在框架内实际实现此异步IO。 Green threads也是一个非常相关的概念，很容易理解。绿色线程和协程是阴影事件循环的一种，它试图实现相同的目的-减少线程，因为我们可以在绿色线程等待某些东西的同时重用系统线程。

How does it even solve c10k problem, when it can't start all concurrent requests parallel and have to wait till the previous one terminates?

当然，我们不会在主线程中等待发送上一个请求的响应。获取请求，安排长时间的/IO任务执行，下一个请求。

Even if I decide to push all these operations to a worker thread(pooled), then I'm back to the same problem isn't it? Context switching between threads?

如果您把所有事情都做对了-不。甚至，您将获得良好的数据局部性和执行流预测。一个CPU内核将执行您的短事件循环并安排异步工作，而无需上下文切换，仅此而已。其他内核调用数据库并仅返回响应。在回调之间切换或检查不同 channel 的IO状态实际上不需要任何系统线程的上下文切换-它实际上在一个工作线程中工作。因此，我们每个核心只有一个工作线程，而这个系统线程正在等待/检查从多个连接到数据库的结果可用性。重新访问 Java NIO概念以了解它如何以这种方式工作。 (NIO的经典示例-代理服务器可以接受许多并行连接(成千上万个)，对其他远程服务器的代理请求，侦听响应并将响应发送回客户端，以及使用一个或两个线程将所有这些发送回客户端)

关于您的代码，我为您制作了一个示例 project，以证明一切正常。

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();
                            })
        );
    }
  }
}

输出:

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

因此，我们接受一个请求-将请求调度到数据库，转到下一个请求，使用所有请求，并仅在数据库完成所有操作后才为每个请求发送响应。

关于您的代码示例，我看到两个可能的问题-首先，看起来您没有 close()连接，这对于将其返回到池很重要。其次，您的池如何配置？如果只有一个空闲连接-这些请求将序列化以等待该连接。

我建议您为两个请求都添加一些时间戳记，以查找序列化的位置。您有一些东西使事件循环中的调用被阻塞。或者...检查您是否在测试中并行发送请求。得到上一个的答复后没有下一个。