2024-4-18 群讨论：关于异步HttpClient如何

发表于

以下来自本人拉的一个关于 Java 技术的讨论群。关注公众号：hashcon，私信进群拉你

群友问题：群友想尽量快的将请求发到三方接口，不考虑三方接口的压力。如何开发并验证？

思路：

  1. 肯定要使用 WebClient 这种异步非阻塞的 io 接口。或者 vertx 这种框架的。虚拟线程先不考虑，因为截止目前不推荐虚拟线程上生产：
    zhuanlan.zhihu.com/p/685013298
  2. 首先的你如果只考虑你的代码的性能，不考虑调用的接口的响应程度，只想尽量把压力吐给他。你不能在测试代码直接调用三方接口，需要先隔离开，先保证你的代码到达你想要的效果。因为这个响应时间，影响因素太多了，并且不稳定。比如你和三方接口之间的带宽，你的网卡，对方是否有限流，以及如果你不限制链接数量，会被 cdn 就拦截限流了。这些都不知道。比如受限于带宽，假设你非阻塞同时发 10000 个请求，那么其实很多是在网络排队等着发。
  3. 只测自己代码，并且想模拟延迟返回。一般测试本地。我一般用 TestContainers 包裹 httpbin 镜像（kennethreitz/httpbin:latest），针对你的场景你可以在每个请求加一个调用时间然后调用 /anything 接口之后收集响应，anything 接口就是返回你发的所有参数。如果你想模拟带宽，可以再加上 toxicproxy 镜像，通过 toxicproxy 去访问 httpbin 去做一定的带宽限流。如果你想模拟接口延迟，可以通过 /delay/0.1 （延迟100ms）实现。

一个代码示例（简单测试，没有仔细调参，只是为了展示测试思路），首先编写 TestContainer 基类，用于复用(这里并不是所有的都用的上，我放出来只是供大家以后测试各种场景)：

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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
ini复制代码import eu.rekawek.toxiproxy.Proxy;
import eu.rekawek.toxiproxy.ToxiproxyClient;
import eu.rekawek.toxiproxy.model.ToxicDirection;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.ToxiproxyContainer;
import org.testcontainers.junit.jupiter.Testcontainers;

import java.io.IOException;

@Testcontainers
public class CommonMicroServiceTest {

    private static final Network network = Network.newNetwork();

    private static final String HTTPBIN = "httpbin";
    public static final int HTTPBIN_PORT = 80;
    public static final GenericContainer<?> HTTPBIN_CONTAINER
            = new GenericContainer<>("kennethreitz/httpbin:latest")
            .withExposedPorts(HTTPBIN_PORT)
            .withNetwork(network)
            .withNetworkAliases(HTTPBIN);
    /**
     * <a href="https://java.testcontainers.org/modules/toxiproxy/">toxiproxy</a>
     * 使用 toxiproxy 封装 httpbin
     * 可以使用 toxiproxy 模拟网络故障等情况
     * 可以用的 port 范围是 8666～8697
     */
    private static final ToxiproxyContainer TOXIPROXY_CONTAINER = new ToxiproxyContainer("ghcr.io/shopify/toxiproxy:2.5.0")
            .withNetwork(network);

    private static final int GOOD_HTTPBIN_PROXY_PORT = 8666;
    private static final int READ_TIMEOUT_HTTPBIN_PROXY_PORT = 8667;
    private static final int RESET_PEER_HTTPBIN_PROXY_PORT = 8668;

    public static final String GOOD_HOST;
    public static final int GOOD_PORT;
    /**
     * 以下代表请求已经发出到服务端，但是响应超时，或者不能响应（比如服务器重启）
     */
    public static final String READ_TIMEOUT_HOST;
    public static final int READ_TIMEOUT_PORT;
    public static final String RESET_PEER_HOST;
    public static final int RESET_PEER_PORT;

    /**
     * 以下代表请求都没有发出去，TCP 链接都没有建立
     */
    public static final String CONNECT_TIMEOUT_HOST = "localhost";
    /**
     * 端口 1 一定连不上的
     */
    public static final int CONNECT_TIMEOUT_PORT = 1;


    static {
        //不使用 @Container 注解管理容器声明周期，因为我们需要在静态块生成代理，必须在这之前启动容器
        //不用担心容器不会被关闭，因为 testcontainers 会启动一个 ryuk 容器，用于监控并关闭所有容器
        HTTPBIN_CONTAINER.start();
        TOXIPROXY_CONTAINER.start();
        final ToxiproxyClient toxiproxyClient = new ToxiproxyClient(TOXIPROXY_CONTAINER.getHost(), TOXIPROXY_CONTAINER.getControlPort());
        try {
            Proxy proxy = toxiproxyClient.createProxy("good", "0.0.0.0:" + GOOD_HTTPBIN_PROXY_PORT, HTTPBIN + ":" + HTTPBIN_PORT);
            //关闭流量，会 READ TIME OUT
            proxy = toxiproxyClient.createProxy("read_timeout", "0.0.0.0:" + READ_TIMEOUT_HTTPBIN_PROXY_PORT, HTTPBIN + ":" + HTTPBIN_PORT);
            proxy.toxics().bandwidth("UP_DISABLE", ToxicDirection.UPSTREAM, 0);
            proxy.toxics().bandwidth("DOWN_DISABLE", ToxicDirection.DOWNSTREAM, 0);
            proxy = toxiproxyClient.createProxy("connect_timeout", "0.0.0.0:" + RESET_PEER_HTTPBIN_PROXY_PORT, HTTPBIN + ":" + HTTPBIN_PORT);
            proxy.toxics().resetPeer("UP_SLOW_CLOSE", ToxicDirection.UPSTREAM, 1);
            proxy.toxics().resetPeer("DOWN_SLOW_CLOSE", ToxicDirection.DOWNSTREAM, 1);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        GOOD_HOST = TOXIPROXY_CONTAINER.getHost();
        GOOD_PORT = TOXIPROXY_CONTAINER.getMappedPort(GOOD_HTTPBIN_PROXY_PORT);
        READ_TIMEOUT_HOST = TOXIPROXY_CONTAINER.getHost();
        READ_TIMEOUT_PORT = TOXIPROXY_CONTAINER.getMappedPort(READ_TIMEOUT_HTTPBIN_PROXY_PORT);
        RESET_PEER_HOST = TOXIPROXY_CONTAINER.getHost();
        RESET_PEER_PORT = TOXIPROXY_CONTAINER.getMappedPort(RESET_PEER_HTTPBIN_PROXY_PORT);
    }
}

测试代码：

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
26
ini复制代码@Test
public void test() {
    // 创建一个自定义的连接提供者
    ConnectionProvider provider = ConnectionProvider.builder("customConnectionProvider")
            .maxConnections(100) // 增加最大连接数，这个不能太大，否则会被 cloudflare 等 cdn 认为是类 ddos 攻击
            .pendingAcquireMaxCount(10000) // 增加等待队列的大小
            .build();

    HttpClient httpClient = HttpClient.create(provider)
            .responseTimeout(Duration.ofMillis(100000)); // 响应超时时间
    WebClient build = WebClient.builder().clientConnector(new ReactorClientHttpConnector(httpClient)).build();
    List<Mono<String>> monos = Lists.newArrayList();
    for (int i = 0; i < 10000; i++) {
        //我这里使用了 TestContainer 的端口模拟 0.1s 延迟
        Mono<String> stringMono = build.get().uri("http://localhost:" +
                CommonMicroServiceTest.HTTPBIN_CONTAINER.getMappedPort(HTTPBIN_PORT) + "/delay/0.1")
                .retrieve().bodyToMono(String.class);
        monos.add(stringMono);
    }
    long start = System.currentTimeMillis();
    String block = Mono.zip(monos, objects -> {
        log.info("{}", objects);
        return "ok";
    }).block();
    log.info("block: {} in {}ms", block, System.currentTimeMillis() - start);
}

测试输出：block: ok in 10362ms

基本符合预期：

  1. 10000 个请求，每个请求 0.1s，链接池 100
  2. 耗时约等于：0.1*10000/100=10s

另外，我一般用 toxicproxy 模拟服务端断开链接，请求发到服务端但是服务端无法响应，请求发不到服务端，发一半到服务端之后的发不到等等等等，在编写微服务基础框架的时候非常好用～

个人简介：个人业余研究了 AI LLM 微调与 RAG，目前成果是微调了三个模型：

  1. 一个模型是基于 whisper 模型的微调，使用我原来做的精翻的视频按照语句段落切分的片段，并尝试按照方言类别，以及技术类别分别尝试微调的成果。用于视频字幕识别。
  2. 一个模型是基于 Mistral Large 的模型的微调，识别提取视频课件的片段，辅以实际的课件文字进行识别微调。用于识别课件的片段。
  3. 最后一个模型是基于 Claude 3 的模型微调，使用我之前制作的翻译字幕，与 AWS、Go 社区、CNCF 生态里面的官方英文文档以及中文文档作为语料，按照内容段交叉拆分，进行微调，用于字幕翻译。

目前，准确率已经非常高了。大家如果有想要我制作的视频，欢迎关注留言。

本人也是开源代码爱好者，贡献过很多项目的源码（Mycat 和 Java JFRUnit 的核心贡献者，贡献过 OpenJDK，Spring，Spring Cloud，Apache Bookkeeper，Apache RocketMQ，Ribbon，Lettuce、 SocketIO、Longchain4j 等项目 ），同时也是深度技术迷，编写过很多硬核的原理分析系列（JVM）。本人也有一个 Java 技术交流群，感兴趣的欢迎关注。

另外，一如即往的是，全网的所有收益，都会捐赠给希望工程，坚持靠爱与兴趣发电。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%