上一章初步介绍了 WRK 基于的事件驱动器 AE，这一章主要介绍 WRK 各命令参数代表的具体含义以及是如何根据参数执行压测的

参数含义

wrk 一般的执行命令：./wrk -t 2 -c 100 -d 10 http://xxxx
其中 t 为线程数（默认 2），c 为总 socket 连接数（默认 10，每个线程数默认的连接数为 10/2 = 5），d 为执行持续的时间（默认 10 秒）

主逻辑

WRK 创建线程：

根据参数 t 创建线程数，可见线程数（t）的大小和实际的并发量没有直接关系，有直接关系的是连接数 t->connection，即是参数 c

for (uint64_t i = 0; i < cfg.threads; i++) { // 这里是通过参数 t 循环创建线程
    thread *t      = &threads[i];
    t->loop        = aeCreateEventLoop(10 + cfg.connections * 3); //aeEventLoop *aeCreateEventLoop(int setsize); 这里创建eventloop,每个线程可监听10+c*3个socket句柄
    t->connections = cfg.connections / cfg.threads; // 每个线程维护的连接数
    if (!t->loop || pthread_create(&t->thread, NULL, &thread_main, t)) {
    }
}

sleep(cfg.duration);  // 通过sleep 一段时间（参数d），将全局stop设为1
stop = 1;

WRK 每条子线程处理逻辑：

为每个线程创建链接，并注册定时器记录结果，并通过 aeMain 阻塞

    for (uint64_t i = 0; i < thread->connections; i++, c++) { // 每个子线程创建 c/t 个socket连接
        c->thread = thread;
        connect_socket(thread, c); // 这里会真正创建连接
    }

aeCreateTimeEvent(loop, RECORD_INTERVAL_MS, record_rate, thread, NULL); // 注册定时器，100ms后执行

aeMain(loop); // 相当于子线程在这里阻塞，直至主动触发 aeStop
aeDeleteEventLoop(loop);

创建链接主逻辑：

fd = socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
// ai_family：AF_UNSPEC，所以wrk原理上支持对ipv6的地址进行压测
// ai_socktype：SOCK_STREAM，说明wrk是基于TCP的
flags = fcntl(fd, F_GETFL, 0);
fcntl(fd, F_SETFL, flags | O_NONBLOCK);  // 证明wrk的socket是非阻塞的，即当无数据可读时会立即返回而不会造成线程阻塞
if (connect(fd, addr->ai_addr, addr->ai_addrlen) == -1) {  // 向目标地址发起连接
    if (errno != EINPROGRESS) goto error;
}
flags = AE_READABLE | AE_WRITABLE;
if (aeCreateFileEvent(loop, fd, flags, socket_connected, c) == AE_OK) { // 针对socket句柄的读写事件注册 socket_connected
}

socket_conneted 里同时注册AE_READABLE 和 AE_WRITABLE
aeCreateFileEvent(c->thread->loop, fd, AE_READABLE, socket_readable, c);
aeCreateFileEvent(c->thread->loop, fd, AE_WRITABLE, socket_writeable, c);
// 如果一个套接字同时产生了这两种事件，那么事件分派器会优先处理AE_READABLE事件，等到AE_READABLE事件处理完之后，才处理AE_WRITABLE事件

socket_writeable：对压测地址发送请求
switch (sock.write(c, buf, len, &n)) { // 向目标地址发送请求
    case OK:    break;
    case ERROR: goto error;
    case RETRY: return;
}

c->written += n;
if (c->written == c->length) {
    c->written = 0;
    aeDeleteFileEvent(loop, fd, AE_WRITABLE); // 发送完成后会删除AE_WRITABLE事件
}

socket_readable：接收压测对象的返回信息
do {
    switch (sock.read(c, &n)) {
        case OK:    break;
        case ERROR: goto error;
        case RETRY: return;
    }

    if (http_parser_execute(&c->parser, &parser_settings, c->buf, n) != n) goto error;
    if (n == 0 && !http_body_is_final(&c->parser)) goto error;

    c->thread->bytes += n;
} while (n == RECVBUF && sock.readable(c) > 0); 
return;

response_complete：wrk会向http_parse注册一个回调on_message_complete，会在服务端响应成功后记录本次请求结果，并且重新注册新的AE_WRITABLE事件
if (!stats_record(statistics.latency, now - c->start)) { // now -c->start 就是请求 - 响应的时延
    thread->errors.timeout++;
}
c->delayed = cfg.delay;
aeCreateFileEvent(thread->loop, c->fd, AE_WRITABLE, socket_writeable, c);

从上面可以看出，

• wrk 会同时对每条 socket 句柄的 AE_WRITABLE，AE_WRITABLE 事件进行注册，在执行 send 后删除 AE_WRITABLE 事件，在 on_message_complete 后重新注册 AE_WRITABLE 事件，所以 wrk 每个线程针对链接的处理时序是 send -> recv -> record -> send -> recv -> record 不断循环
• wrk 计算延迟的方式是通过 response_time - request_time，所以无法避免性能压测中的 Coordinated Omission 问题（在 wrk2 中有解决）

定时记录逻辑：

record_rate:
    if (thread->requests > 0) {
        uint64_t elapsed_ms = (time_us() - thread->start) / 1000;
        uint64_t requests = (thread->requests / (double) elapsed_ms) * 1000;

        stats_record(statistics.requests, requests);

        thread->requests = 0;
        thread->start    = time_us();
    }

    if (stop) aeStop(loop); // 当主线程将stop设为1时，子线程停止阻塞并结束

功能扩展

集合点：

wrk 的 connection 虽然理解上为并发量，但是由于每个连接的响应不一致，无法保障同一时间点若干连接同时发起请求（类似 JMETER/LR 的集合点）

具体思路：

• 在 socket_writeable 中获取当前待发送数据的 socket 句柄数
• 当满足预期值时，则发送
• 当不满足预期值时，删除当前 AE_WRITABLE 事件，创建定时器等待，并重新注册 AE_WRITABLE