JMH Microbenchamrks (conclusions from tests - channel caching improved Sender#send and destroyed Sender#sendWithPublishConfirms)

Author: flooq

build.gradle

@@ -25,6 +25,7 @@ buildscript {
 }
 plugins {
 	id 'org.asciidoctor.convert' version '1.5.6'
+	id 'me.champeau.gradle.jmh' version '0.4.7'
 }
 
 ext {

src/jmh/java/reactor/rabbitmq/SenderBenchmark.java

@@ -0,0 +1,61 @@
+package reactor.rabbitmq;
+
+import com.rabbitmq.client.Connection;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+import reactor.core.publisher.Flux;
+
+import java.util.concurrent.TimeUnit;
+
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Benchmark)
+@Warmup(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 1, time = 5, timeUnit = TimeUnit.SECONDS)
+@Fork(1)
+@Threads(2)
+public class SenderBenchmark {
+
+    Connection connection;
+    Sender sender;
+    String queue;
+    Flux<OutboundMessage> msgFlux;
+
+    @Param({"1", "10", "100", "1000"})
+    public int nbMessages;
+
+    @Setup
+    public void setupConnection() throws Exception {
+        connection = SenderBenchmarkUtils.newConnection();
+    }
+
+    @TearDown
+    public void closeConnection() throws Exception {
+        connection.close();
+    }
+
+    @Setup(Level.Iteration)
+    public void setupSender() throws Exception {
+        queue = SenderBenchmarkUtils.declareQueue(connection);
+        sender = RabbitFlux.createSender();
+        msgFlux = SenderBenchmarkUtils.outboundMessageFlux(queue, nbMessages);
+    }
+
+    @TearDown(Level.Iteration)
+    public void tearDownSender() throws Exception {
+        SenderBenchmarkUtils.deleteQueue(connection, queue);
+        if (sender != null) {
+            sender.close();
+        }
+    }
+
+    @Benchmark
+    public void send(Blackhole blackhole) {
+        blackhole.consume(sender.send(msgFlux).block());
+    }
+
+    @Benchmark
+    public void sendWithPublishConfirms(Blackhole blackhole) {
+        blackhole.consume(sender.sendWithPublishConfirms(msgFlux).blockLast());
+    }
+}

src/jmh/java/reactor/rabbitmq/SenderBenchmarkUtils.java

@@ -0,0 +1,36 @@
+package reactor.rabbitmq;
+
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import reactor.core.publisher.Flux;
+
+import java.util.UUID;
+
+public class SenderBenchmarkUtils {
+
+    public static Flux<OutboundMessage> outboundMessageFlux(String queue, int nbMessages) {
+        return Flux.range(0, nbMessages).map(i -> new OutboundMessage("", queue, "".getBytes()));
+    }
+
+    public static Connection newConnection() throws Exception {
+        ConnectionFactory connectionFactory = new ConnectionFactory();
+        connectionFactory.useNio();
+        return connectionFactory.newConnection();
+    }
+
+    public static String declareQueue(Connection connection) throws Exception {
+        String queueName = UUID.randomUUID().toString();
+        Channel channel = connection.createChannel();
+        String queue = channel.queueDeclare(queueName, false, false, false, null).getQueue();
+        channel.close();
+        return queue;
+    }
+
+    public static void deleteQueue(Connection connection, String queue) throws Exception {
+        Channel channel = connection.createChannel();
+        channel.queueDelete(queue);
+        channel.close();
+    }
+
+}

src/jmh/java/reactor/rabbitmq/SenderWithLazyChannelPoolBenchmark.java

@@ -0,0 +1,69 @@
+package reactor.rabbitmq;
+
+import com.rabbitmq.client.Connection;
+import org.openjdk.jmh.annotations.*;
+import org.openjdk.jmh.infra.Blackhole;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.concurrent.TimeUnit;
+
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@State(Scope.Benchmark)
+@Warmup(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 1, time = 5, timeUnit = TimeUnit.SECONDS)
+@Fork(1)
+@Threads(2)
+public class SenderWithLazyChannelPoolBenchmark {
+
+    Connection connection;
+    ChannelPool channelPool;
+    Sender sender;
+    String queue;
+    Flux<OutboundMessage> msgFlux;
+
+    @Param({"1", "10", "25"})
+    public int channelPoolSize;
+
+    @Param({"1", "10", "100", "1000"})
+    public int nbMessages;
+
+    @Setup
+    public void setupConnection() throws Exception {
+        connection = SenderBenchmarkUtils.newConnection();
+        channelPool = ChannelPoolFactory.createChannelPool(Mono.just(connection), new ChannelPoolOptions().maxCacheSize(channelPoolSize));
+    }
+
+    @TearDown
+    public void closeConnection() throws Exception {
+        connection.close();
+        channelPool.close();
+    }
+
+    @Setup(Level.Iteration)
+    public void setupSender() throws Exception {
+        queue = SenderBenchmarkUtils.declareQueue(connection);
+        sender = RabbitFlux.createSender();
+        msgFlux = SenderBenchmarkUtils.outboundMessageFlux(queue, nbMessages);
+    }
+
+    @TearDown(Level.Iteration)
+    public void tearDownSender() throws Exception {
+        SenderBenchmarkUtils.deleteQueue(connection, queue);
+        if (sender != null) {
+            sender.close();
+        }
+    }
+
+//    @Benchmark
+    public void send(Blackhole blackhole) {
+        blackhole.consume(sender.send(msgFlux, new SendOptions().channelPool(channelPool)).block());
+    }
+
+//    @Benchmark
+    public void sendWithPublishConfirms(Blackhole blackhole) {
+        blackhole.consume(sender.sendWithPublishConfirms(msgFlux, new SendOptions().channelPool(channelPool)).blockLast());
+    }
+
+}