computer science►engineering architecture►rpc

brpc

自定义协议

示例

1
2
3
4
5
6
7
8
9
10
Protocol protocol = { ParseMessage,
                      SerializeRequest, PackRequest,
                      ProcessHttpRequest, ProcessResponse,
                      VerifyRequest, ParseServerAddress,
                      GetMethodName,
                      CONNECTION_TYPE_SINGLE,
                      "h2" };
if (RegisterProtocol(PROTOCOL_H2, protocol) != 0) {
    exit(1);
}

客户端调用顺序

  • 去程
    • SerializeRequest
    • PackRequest
  • 回程
    • ParseMessage
    • ProcessResponse

bthread

初始化

1
2
3
4
5
6
7
8
9
10
bthread_t bid;
bthread_attr_t attr;
bthread_attr_init(&attr);

auto s = [](void *) -> void * {
    auto r = bthread_usleep(1000 * 1000 * 3);
    return nullptr;
};
bthread_start_urgent(&bid, &attr, s, nullptr);
bthread_join(bid, nullptr);

使用

1
2
3
4
5
6
7
8
#include "bthread/bthread.h"
auto s = [](void *) -> void * {
    bthread_usleep(1000 * 15);
};

bthread_t th1;
bthread_start_urgent(&th1, &attr, s, nullptr);
bthread_join(th1, nullptr);

timer_thread

NOTE: 执行一次，且时间为时间戳（通常是未来时间），非间隔执行。

1
2
3
4
5
6
#include "bthread/timer_thread.h"
bthread::TimerThread timer;
bthread::TimerThreadOptions options;
timer.start(&options); 
// 或者 timer.start(nullptr); 
timer.schedule(ConfigData::load, this, {butil::seconds_from_now{5}}); // 5秒后执行一次

自定义 Period Timer

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
// brpc_period_timer.h
typedef void (*BrpcPeriodTimerFunction)(void *);

struct BrpcTimerContext {
    BrpcPeriodTimerFunction function;
    void *argument;
    int interval_s;
    bthread::TimerThread *timer;
};

class BrpcPeriodTimer {
public:
    BrpcPeriodTimer();
    ~BrpcPeriodTimer();

    static void Wrapper(void *ctx);
    void Schedule(BrpcTimerContext *);
private:
    bthread::TimerThread timer_;
};

// brpc_period_timer.cpp
BrpcPeriodTimer::BrpcPeriodTimer() {
    timer_.start(nullptr);
}

BrpcPeriodTimer::~BrpcPeriodTimer() {
    timer_.stop_and_join();
}

void BrpcPeriodTimer::Wrapper(void *ctx) {
    auto c = (BrpcTimerContext *) ctx;
    c->function(c->argument);
    c->timer->schedule(BrpcPeriodTimer::Wrapper, ctx, butil::seconds_from_now(c->interval_s));
}

void BrpcPeriodTimer::Schedule(BrpcTimerContext *ctx) {
    auto c = (BrpcTimerContext *) ctx;
    ctx->timer = &timer_;
    timer_.schedule(BrpcPeriodTimer::Wrapper, ctx, butil::seconds_from_now(c->interval_s));
}

// test
TEST(Utils, PeriodTimer) {
    auto payload = [](void *) {
        std::cout << "run payload" << std::endl;
    };

    BrpcPeriodTimer timer;
    BrpcTimerContext ctx{payload, nullptr, 1};
    timer.Schedule(&ctx);

    sleep(10);
}

// output
Testing started at 17:45 ...
run payload
run payload
run payload
run payload
run payload
run payload
run payload
run payload
run payload
Process finished with exit code 0