[Feature Request] Support executing callbacks concurrently

Aberta
  • Algo errado?
#368 0 comentários 0 reações 0 responsáveis Ver no GitHub

Ninguém assumiu esta issue ainda.

Avaliação

Dificuldade
5/5
Tempo estimado
Mais de uma semana
Facilidade para iniciantes
35/100
Tipo de issue
Funcionalidade
Clareza
Razoavelmente clara
Status de atividade
Estagnada
Stack de tecnologia
cpp
Domínio
api

Direção de pesquisa

Comece pelo caminho de callback de createProducerAsync e pelo tratamento de conexões do event-loop descritos no relatório; em seguida, compare o comportamento do executor de CompletableFuture thenXxxAsync do cliente Java. O trabalho estará concluído quando os callbacks puderem ser executados simultaneamente sem bloquear outras operações assíncronas e os usuários tiverem uma forma de controlar a execução dos callbacks.

Escrita pelo modelo de indexação a partir do texto da issue.

Descrição

#include <pulsar/Client.h>

#include <chrono>
#include <thread>

#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 | [<none> -> 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 | [<none> -> 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.

Linguagem predominante
C++
Estrelas
72
Forks
93
Nenhum PR com merge em 30d

Preparar o ambiente

Primeiros passos

  1. Leia a issue inteira e depois o guia de contribuição do projeto.
  2. Comente na issue dizendo que vai assumir — evita que duas pessoas façam o mesmo trabalho.
  3. Faça um fork do repositório e trabalhe em uma branch.
  4. Abra um pull request que referencie o número da issue.

Mais de apache/pulsar-client-cpp

Todas as issues de apache/pulsar-client-cpp

Issues semelhantes

Mais issues de C++

Receba novas issues na sua caixa de entrada

Um resumo curto de issues do GitHub para quem está começando.