apache / apache/pulsar-client-cpp

[Feature Request] Support executing callbacks concurrently

Đang mở
#368 0 bình luận 0 reaction 0 người được giao Xem trên GitHub
Ngôn ngữ chính
C++
Star
71
Fork
90
Merge trung bình
2 giờ 33 phút
Pull request đã merge (30 ngày)
3

Mô tả

```c++
#include

#include
#include

#include "lib/LogUtils.h"

DECLARE_LOG_OBJECT()
using namespace pulsar;

int main(int argc, char *argv[]) {
ClientConfiguration conf;
conf.setIOThreads(8);
Client client{"pulsar://localhost:6650", conf};
for (int i = 0; i < 5; i++) {
auto topic = "my-topic-" + std::to_string(i);
LOG_INFO(" XYZ Before create producer for " << topic);
client.createProducerAsync(topic, [topic](Result result, Producer producer) {
LOG_INFO(" XYZ After create producer for " << topic << ": " << result);
producer.sendAsync(MessageBuilder().setContent("msg").build(),
[](Result result, const MessageId &msgId) {
LOG_INFO("XYZ send: " << result << ", " << msgId);
});
std::this_thread::sleep_for(std::chrono::hours(1));
});
}
std::this_thread::sleep_for(std::chrono::hours(1));
}
```

Output:

```
2023-12-06 20:01:08.044 INFO [0x1fb6f1e00] SampleProducer:18 | XYZ Before create producer for my-topic-0
2023-12-06 20:01:08.130 INFO [0x1fb6f1e00] ClientConnection:190 | [ -> pulsar://localhost:6650] Create ClientConnection, timeout=10000
2023-12-06 20:01:08.130 INFO [0x1fb6f1e00] ConnectionPool:114 | Created connection for pulsar://localhost:6650-0
2023-12-06 20:01:08.143 INFO [0x1fb6f1e00] SampleProducer:18 | XYZ Before create producer for my-topic-1
2023-12-06 20:01:08.143 INFO [0x1fb6f1e00] SampleProducer:18 | XYZ Before create producer for my-topic-2
2023-12-06 20:01:08.143 INFO [0x1fb6f1e00] SampleProducer:18 | XYZ Before create producer for my-topic-3
2023-12-06 20:01:08.144 INFO [0x1fb6f1e00] SampleProducer:18 | XYZ Before create producer for my-topic-4
2023-12-06 20:01:08.163 INFO [0x16ef87000] ClientConnection:404 | [127.0.0.1:54469 -> 127.0.0.1:6650] Connected to broker
2023-12-06 20:01:08.344 INFO [0x16ef87000] HandlerBase:83 | [persistent://public/default/my-topic-0, ] Getting connection from pool
2023-12-06 20:01:08.347 INFO [0x16ef87000] HandlerBase:83 | [persistent://public/default/my-topic-2, ] Getting connection from pool
2023-12-06 20:01:08.347 INFO [0x16ef87000] HandlerBase:83 | [persistent://public/default/my-topic-1, ] Getting connection from pool
2023-12-06 20:01:08.347 INFO [0x16ef87000] HandlerBase:83 | [persistent://public/default/my-topic-4, ] Getting connection from pool
2023-12-06 20:01:08.347 INFO [0x16ef87000] HandlerBase:83 | [persistent://public/default/my-topic-3, ] Getting connection from pool
2023-12-06 20:01:08.355 INFO [0x16ef87000] BinaryProtoLookupService:87 | Lookup response for persistent://public/default/my-topic-0, lookup-broker-url pulsar://127.0.0.1:6650, from [127.0.0.1:54469 -> 127.0.0.1:6650]
2023-12-06 20:01:08.355 INFO [0x16ef87000] ClientConnection:190 | [ -> pulsar://localhost:6650] Create ClientConnection, timeout=10000
2023-12-06 20:01:08.355 INFO [0x16ef87000] ConnectionPool:114 | Created connection for pulsar://127.0.0.1:6650-0
2023-12-06 20:01:08.357 INFO [0x16ef87000] BinaryProtoLookupService:87 | Lookup response for persistent://public/default/my-topic-2, lookup-broker-url pulsar://127.0.0.1:6650, from [127.0.0.1:54469 -> 127.0.0.1:6650]
2023-12-06 20:01:08.357 INFO [0x16ef87000] BinaryProtoLookupService:87 | Lookup response for persistent://public/default/my-topic-1, lookup-broker-url pulsar://127.0.0.1:6650, from [127.0.0.1:54469 -> 127.0.0.1:6650]
2023-12-06 20:01:08.357 INFO [0x16ef87000] BinaryProtoLookupService:87 | Lookup response for persistent://public/default/my-topic-4, lookup-broker-url pulsar://127.0.0.1:6650, from [127.0.0.1:54469 -> 127.0.0.1:6650]
2023-12-06 20:01:08.357 INFO [0x16ef87000] BinaryProtoLookupService:87 | Lookup response for persistent://public/default/my-topic-3, lookup-broker-url pulsar://127.0.0.1:6650, from [127.0.0.1:54469 -> 127.0.0.1:6650]
2023-12-06 20:01:08.363 INFO [0x16ef87000] ClientConnection:406 | [127.0.0.1:54470 -> 127.0.0.1:6650] Connected to broker through proxy. Logical broker: pulsar://127.0.0.1:6650
2023-12-06 20:01:08.427 INFO [0x16ef87000] ProducerImpl:212 | [persistent://public/default/my-topic-2, ] Created producer on broker [127.0.0.1:54470 -> 127.0.0.1:6650]
2023-12-06 20:01:08.428 INFO [0x16ef87000] SampleProducer:20 | XYZ After create producer for my-topic-2: Ok
```

Only one producer was created successfully. Other producers were blocked because all callbacks of `createProducerAsync` were executed in the same I/O thread, even though the number of I/O threads is 8.

That's because each event loop of a connection to broker uses the same thread to execute callbacks. So if a callback is blocked, other asynchronous calls will be blocked. Increasing `connectionsPerBroker` can make more I/O threads be used, but users cannot control which thread the callback is executed.

In the Java client, the asynchronous API returns a `CompletableFuture` whose `thenXxxAsync` method can execute the callback in a user provided executor.

Hướng dẫn đóng góp

Mở hướng dẫn đóng góp

Hướng nghiên cứu

Bắt đầu với đường dẫn callback của createProducerAsync và việc xử lý kết nối của event-loop được mô tả trong báo cáo, sau đó so sánh hành vi executor của CompletableFuture thenXxxAsync trong Java client. Công việc hoàn tất khi các callback có thể được thực thi đồng thời mà không chặn các thao tác bất đồng bộ khác và người dùng có cách kiểm soát việc thực thi callback.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Đánh giá

Công nghệ
cpp
Lĩnh vực
api
Loại issue
Tính năng
Độ khó
5/5
Thời gian dự kiến
Hơn một tuần
Mức độ hoạt động
Đình trệ
Độ rõ ràng
Khá rõ ràng
Mức phù hợp với người mới
35/100

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.