apache / apache/pulsar-client-cpp
[Feature Request] Support executing callbacks concurrently
- 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
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