Quarkus Kafka - потребитель пакетных / массовых сообщений

Я хочу пакетный процесс. В моем случае отправки сообщения производителя kafka отправляются одно за другим. Я хочу прочитать их как список в потребительском приложении. Я могу сделать это в библиотеке Spring Kafka. пакетный прослушиватель Spring Kafka

Есть ли способ сделать это с помощью библиотеки quarkus-smallrye-reactive-messaging-kafka?

Я попробовал приведенный ниже пример, но получил ошибку.

ERROR [io.sma.rea.mes.provider] (vert.x-eventloop-thread-3) SRMSG00200: The method org.MyConsumer#aggregate has thrown an exception: java.lang.ClassCastException: class org.TestConsumer cannot be cast to class io.smallrye.mutiny.Multi (org.TestConsumer is in unnamed module of loader io.quarkus.bootstrap.classloading.QuarkusClassLoader @6f2c0754; io.smallrye.mutiny.Multi is in unnamed module of loader io.quarkus.bootstrap.classloading.QuarkusClassLoader @4c1638b)

application.properties:

kafka.bootstrap.servers=hosts
mp.messaging.connector.smallrye-kafka.group.id=KafkaQuick
mp.messaging.connector.smallrye-kafka.auto.offset.reset=earliest
mp.messaging.incoming.test-consumer.connector=smallrye-kafka
mp.messaging.incoming.test-consumer.value.deserializer=org.TestConsumerDeserializer

TestConsumerDeserializer:

public class TestConsumerDeserializer extends JsonbDeserializer<TestConsumer>{
    public TestConsumerDeserializer(){
         // pass the class to the parent.
         super(TestConsumer.class);
    }
}

MyConsumer:

@ApplicationScoped
public class MyConsumer {
    
    @Incoming("test-consumer")
    //@Outgoing("aggregated-channel")
    public void aggregate(Multi<Message<TestConsumer>> in) {
        System.out.println(in);
    }
}

apache-kafka quarkus smallrye-reactive-messaging
person Süleyman Can    schedule 19.01.2021    source источник
comment
Вы можете сами собрать данные в Очередь, а затем обработать / очистить их, когда они достигнут установленного размера.   -  person OneCricketeer    schedule 20.01.2021

Ответы (1)


arrow_upward
0
arrow_downward

Я не понимаю причину возникновения исключения ClassNotFoundException в вопросе. Но я нашел решения для чтения массовых сообщений с использованием quarkus-smallrye-reactive-messaging-kafka.

Решение 1:

@Incoming("test-consumer-topic")
@Outgoing("aggregated-channel")
public Multi<List<TestConsumer>> aggregate(Multi<TestConsumer> in) {
     return in.groupItems().intoLists().every(Duration.ofSeconds(5));
}

@Incoming("aggregated-channel")
public void test(List<TestConsumer> test) {
   System.out.println("size: "+ test.size());
}

Решение 2:

@Incoming("test-consumer-topic")
@Outgoing("events-persisted")
public Multi<Message<TestConsumer>> processPayloadStream(Multi<Message<TestConsumer>> messages) {
    return messages
                    .groupItems().intoLists().of(4)
                    .emitOn(Infrastructure.getDefaultWorkerPool())
                    .flatMap(messages1 -> {
                        persist(messages1);
                        return Multi.createFrom().items(messages1.stream());
                    }).emitOn(Infrastructure.getDefaultExecutor());
}

public void persist(List<Message<TestConsumer>> messages){
    System.out.println("messages size:"+ messages.size());
}

@Incoming("events-persisted")
public CompletionStage<Void> messageAcknowledging(Message<TestConsumer> message){
    return message.ack();
}

примечание: использование конфигурации application.properties в вопросе.

использованная литература:

Поддержка подписки с помощью Multi ‹Message‹ ›› ...

Получение сообщений массового опроса от kafka

person Süleyman Can    schedule 20.01.2021