danielkec

mp.messaging:

  incoming.from-kafka:
    connector: helidon-kafka
    topic: messaging-test-topic-1
    auto.offset.reset: latest
    enable.auto.commit: true

  outgoing.to-kafka:
    connector: helidon-kafka

danielkec

https://github.com/danielkec/helidon-jmh/tree/emitter-lock-comparison/src/main/java/io/helidon

Single thread test

• WIN 10
• AMD Ryzen 7 3700X 8-Core Processor, 3593 Mhz
• JMH version: 1.19
• VM version: JDK 14.0.1, VM 14.0.1+7

Benchmark                                Mode  Cnt   Score   Error  Units

danielkec

package io.helidon.reactive.jmh;

import java.util.List;
import java.util.concurrent.Flow;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import io.helidon.common.reactive.BufferedEmittingPublisher;
import io.helidon.common.reactive.Multi;
import io.helidon.common.reactive.OriginThreadPublisher;

danielkec

Multi.just(1, 2, 3, 4, 5)
       .flatMap(i -> Multi.just(i, i))
       .limit(4)
       .distinct()
       .forEach(System.out::println);

> 1
> 2

danielkec

<dependency>
   <groupId>io.helidon.microprofile.reactive-streams</groupId>
   <artifactId>helidon-microprofile-reactive-streams</artifactId>
   <version>${helidon.version}</version>
</dependency>

danielkec

@Outgoing("int-channel")
public Publisher<Integer> produceInts() {
   return Flowable.just(1, 2, 3);
}

@Incoming("int-channel")
public void consumeInt(Integer value) {
   System.out.println("Consuming Integer: " + value);
}

danielkec

AtomicInteger sum = new AtomicInteger();

Flux<Integer> flux = Flux.just("1", "2", "3", "4", "5")
       .map(Integer::parseInt);

ReactiveStreams.fromPublisher(flux)
       .limit(3)
       .forEach(sum::addAndGet)
       .run()
       .whenComplete((r, t) -> System.out.println("Sum: " + sum.get()));

danielkec

AtomicInteger sum = new AtomicInteger();

Flowable<Integer> flowable = Flowable.just("1", "2", "3", "4", "5")
       .map(Integer::parseInt);

ReactiveStreams.fromPublisher(flowable)
       .limit(3)
       .forEach(sum::addAndGet)
       .run()
       .whenComplete((r, t) -> System.out.println("Sum: " + sum.get()));

danielkec

AtomicInteger sum = new AtomicInteger();

ReactiveStreams.of("1", "2", "3", "4", "5")
       .limit(3)
       .map(Integer::parseInt)
       .forEach(sum::addAndGet)
       .run()
       .whenComplete((r, t) -> System.out.println("Sum: " + sum.get()));

> Sum: 6

danielkec

@Outgoing("test-channel")
public Publisher<Message<String>> produceMessage() {
   return ReactiveStreams.of(Message.of("test-data", () -> {
       System.out.println("Message acked!");
       return CompletableFuture.completedStage(null);
   })).buildRs();
}

@Incoming("test-channel")
@Acknowledgment(Acknowledgment.Strategy.POST_PROCESSING)