假設(shè)我們開(kāi)發(fā)了一個(gè)必須與其他HTTP服務(wù)來(lái)交互的服務(wù)。不幸的是，這些HTTP服務(wù)速度慢且是阻塞的。

它可能是一個(gè)非常慢的遺留HTTP服務(wù)或我們必須使用的一些阻塞 API。無(wú)論如何，我們無(wú)法控制它。在這里，我們將調(diào)用兩個(gè)HTTP API。其中一個(gè)將阻塞2秒鐘，另一個(gè)將阻塞5秒鐘。

一旦兩個(gè)響應(yīng)都可用，我們還需要打印響應(yīng)狀態(tài)代碼。如果我們以老的、非異步反應(yīng)性方式執(zhí)行此操作，我們將阻塞調(diào)用線程5秒鐘。阻塞線程 5 秒效率不高，不是嗎？現(xiàn)在就叫你如何使用 Vert.x 異步發(fā)送HTTP長(zhǎng)阻塞請(qǐng)求來(lái)提高并發(fā)響應(yīng)!

服務(wù)

我使用“httpstat.us”作為網(wǎng)絡(luò)服務(wù)。這是一個(gè)簡(jiǎn)單的服務(wù)，用于生成不同的HTTP代碼來(lái)測(cè)試Web客戶端?？梢蕴峁╊~外的參數(shù)，在本例中為 sleep ，在規(guī)定的時(shí)間內(nèi)阻塞 HTTP 請(qǐng)求。

我將使用“httpie”來(lái)測(cè)試這兩種服務(wù)。

服務(wù) 1 將阻塞5秒鐘，并返回狀態(tài)代碼為 200 的響應(yīng)：

http://httpstat.us/200?sleep=5000
_____________________________________________

HTTP/1.1 200 OK
Content-Length: 6
Content-Type: text/plain
Date: Tue, 08 Mar 2022 17:05:08 GMT
Request-Context: appId=cid-v1:1e93d241-20e4-4513-bbd7-f452a16a5d69
Server: Kestrel
Set-Cookie: ARRAffinity=e2c17206c539113795daf64bd958d003f2b29b9f62da53617beea05468875ba5;Path=/;HttpOnly;Domain=httpstat.us

200 OK

服務(wù) 2 與前一個(gè)相同，只是它阻塞了2秒而不是5秒：

http://httpstat.us/200?sleep=2000
_____________________________________________

HTTP/1.1 200 OK
Content-Length: 6
Content-Type: text/plain
Date: Tue, 08 Mar 2022 17:11:53 GMT
Request-Context: appId=cid-v1:1e93d241-20e4-4513-bbd7-f452a16a5d69
Server: Kestrel
Set-Cookie: ARRAffinity=e2c17206c539113795daf64bd958d003f2b29b9f62da53617beea05468875ba5;Path=/;HttpOnly;Domain=httpstat.us

200 OK

Web 客戶端

我們已經(jīng)了解了服務(wù)。現(xiàn)在，讓我們討論 Web 客戶端?，F(xiàn)在，我將使用 Vert.x Web 客戶端。它是一個(gè)異步的，易于使用的 HTTP 和 HTTP/2 客戶端.

  private static Future<Integer> service1(WebClient webClient) {
    return webClient.getAbs("http://httpstat.us/200?sleep=5000")
        .send()
        .onSuccess(response -> System.out.println(MessageFormat.format("[{0}] service 1: response received", Thread.currentThread().getName())))
        .compose(response -> Future.succeededFuture(response.statusCode()));
  }

  private static Future<Integer> service2(WebClient webClient) {
    return webClient.getAbs("http://httpstat.us/200?sleep=2000")
        .send()
        .onSuccess(response -> System.out.println(MessageFormat.format("[{0}] service 2 response received", Thread.currentThread().getName())))
        .compose(response -> Future.succeededFuture(response.statusCode()));
  }

這兩種方法非常相似。它們將 WebClient 作為參數(shù)并發(fā)送返回 Future<Integer> 的 HTTP 請(qǐng)求。其中整數(shù)是 HTTP 響應(yīng)代碼。返回的 Future<Integer> 向我們保證結(jié)果是異步的。狀態(tài)代碼在稍后可用時(shí)將給出回調(diào)。

我們需要組合兩個(gè)Future。使用 Vert.x的 CompositeFuture 可以實(shí)現(xiàn)多個(gè)future的協(xié)調(diào). 它支持并發(fā)組合(并行運(yùn)行多個(gè)異步操作)和順序組合(鏈?zhǔn)疆惒讲僮?.

    Future<Integer> service1Code = service1(webClient);
    Future<Integer> service2Code = service2(webClient);

    CompositeFuture.all(service1Code, service2Code)
        .onSuccess(ar -> {
          printResult(ar);

          countDownLatch.countDown();
        });

合在一起

最后，我們可以將所有的點(diǎn)點(diǎn)滴滴放在一起，如下所示：

import java.text.MessageFormat;
import java.util.concurrent.CountDownLatch;

import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.ext.web.client.WebClient;

public class Services {
  public static void main(String[] args) throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(1);

    // Vertx instance and web client
    Vertx     vertx     = Vertx.vertx();
    WebClient webClient = WebClient.create(vertx);

    Future<Integer> service1Code = service1(webClient);
    Future<Integer> service2Code = service2(webClient);

    CompositeFuture.all(service1Code, service2Code)
        .onSuccess(ar -> {
          printResult(ar);

          countDownLatch.countDown();
        });

    vertx.setPeriodic(1000, l -> System.out.println("[" + Thread.currentThread().getName() + "] is released"));

    countDownLatch.await();
    vertx.close();
  }

  private static Future<Integer> service1(WebClient webClient) {
    return webClient.getAbs("http://httpstat.us/200?sleep=5000")
        .send()
        .onSuccess(response -> System.out.println(MessageFormat.format("[{0}] service 1: response received", Thread.currentThread().getName())))
        .compose(response -> Future.succeededFuture(response.statusCode()));
  }

  private static Future<Integer> service2(WebClient webClient) {
    return webClient.getAbs("http://httpstat.us/200?sleep=2000")
        .send()
        .onSuccess(response -> System.out.println(MessageFormat.format("[{0}] service 2 response received", Thread.currentThread().getName())))
        .compose(response -> Future.succeededFuture(response.statusCode()));
  }

  private static void printResult(CompositeFuture compositeFuture) {
    System.out.println(Thread.currentThread().getName() + " Result: service1:" + compositeFuture.resultAt(0) + " service2:" + compositeFuture.resultAt(1)
    );
  }
}

下面是運(yùn)行代碼后打印在控制臺(tái)上的結(jié)果。這兩個(gè)請(qǐng)求都是從同一個(gè) vertx 事件循環(huán)線程調(diào)度的。該程序還會(huì)每秒打印線程未被阻止的消息。最后，它將打印兩個(gè)狀態(tài)代碼作為最終結(jié)果。如您所見(jiàn)，一切都發(fā)生在同一個(gè)線程上：

[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] service 2 response received
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] service 1: response received
[vert.x-eventloop-thread-1] Result: service1:200 service2:200

總結(jié)

這就是我目前所要討論的全部?jī)?nèi)容。我希望這篇文章能幫助您現(xiàn)在對(duì)如何使用 Vert.x 異步發(fā)送長(zhǎng)阻塞請(qǐng)求有了更好的理解。文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-438293.html

到了這里，關(guān)于使用 Vert.x 異步發(fā)送HTTP長(zhǎng)阻塞請(qǐng)求來(lái)提高并發(fā)響應(yīng)的文章就介紹完了。如果您還想了解更多內(nèi)容，請(qǐng)?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！