Performance / blocking / non-blocking

Application.java

@SpringBootApplication()
public class Application {
  public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
  }

  @Bean
  @Qualifier("completableFutureWorkerPool")
  public Executor asyncTaskExecutor() {
    return Executors.newCachedThreadPool(new CustomizableThreadFactory("completableFutureWorkerPool-"));
  }
}

BasicSimulation.scala

// This simulation takes 30sec and was run against each endpoint with different 
// parameters in delayMs and different concurrent users (see below, rampUsers)
class BasicSimulation extends Simulation {

  val httpProtocol = http
    .baseUrl("http://localhost:8080") // Here is the root for all relative URLs
    .userAgentHeader("Mozilla/5.0 (Macintosh; Intel Mac OS X 10.8; rv:16.0) Gecko/20100101 Firefox/16.0")

  val scn = scenario("cf-pooled-sleep?delayMs=20") // A scenario is a chain of requests and pauses
    .exec(
      http("request_1")
        .get("/perf/cf-pooled-sleep?delayMs=20")
    )
    .exec(
      http("request_2")
        .get("/perf/cf-pooled-sleep?delayMs=20")
    )
    .exec(
      http("request_3")
        .get("/perf/cf-pooled-sleep?delayMs=20")
    )
    .exec(
      http("request_4")
        .get("/perf/cf-pooled-sleep?delayMs=20")
    )
    .exec(
      http("request_5")
        .get("/perf/cf-pooled-sleep?delayMs=20")
    )

  setUp(scn.inject(rampUsers(10000) during (30 seconds))).maxDuration(1 minutes).protocols(httpProtocol)
}

Javalin.kt

// Similar HTTP API using Javalin with Jetty using a QueuingThreadPool with 200 threads 
// by default. Javalin supports non-blocking as well using CompletableFuture 

package nl.toefel.javalin.performance

import com.fasterxml.jackson.databind.ObjectMapper
import io.javalin.Javalin
import io.javalin.http.Context
import org.slf4j.Logger
import org.slf4j.LoggerFactory
import java.util.concurrent.CompletableFuture
import java.util.concurrent.Executors
import java.util.function.Supplier

fun main() {
    Router(8080).start()
}

data class CustomerDto(
    val id: Int?,
    val firstName: String,
    val lastName: String
)

class Router(private val port: Int) {
    private val logger: Logger = LoggerFactory.getLogger(Router::class.java)
    private val executor = Executors.newCachedThreadPool()
    private val mapper = ObjectMapper().findAndRegisterModules()

    val app = Javalin.create { cfg -> cfg.requestLogger(::logRequest).enableCorsForAllOrigins() }
        .get("/perf/instant", ::getCustomer)
        .get("/perf/future", ::getCustomerFuture)

    private fun logRequest(ctx: Context, executionTimeMs: Float) =
        logger.info("${ctx.method()} ${ctx.fullUrl()} status=${ctx.status()} durationMs=$executionTimeMs")

    fun start(): Router {
        app.start(port)
        return this
    }

    private fun getCustomer(ctx: Context) {
        val delay = ctx.queryParam("delayMs")?.toLong() ?: 0
        if (delay > 0) {
            Thread.sleep(delay)
        }
        ctx.json(CustomerDto(
            id = 1,
            firstName = "TestDto",
            lastName = "Jemeur"
        ))
    }

    private fun getCustomerFuture(ctx: Context) {
        val delay = ctx.queryParam("delayMs")?.toLong() ?: 0L
        val cf = CompletableFuture.supplyAsync(Supplier<String> {
            if (delay > 0) {
                Thread.sleep(delay)
            }
            mapper.writeValueAsString(
                CustomerDto(
                    id = 1,
                    firstName = "TestDto",
                    lastName = "Jemeur"
                ))
        }, executor)
        ctx.result(cf)
    }
}

PerfController.java

package net.intergamma.bar.v1.service;

import lombok.extern.slf4j.Slf4j;
import net.intergamma.bar.v1.domain.Bar;
import net.intergamma.microservice.commons.SpringProfileResolver;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;

// This controller contains different methods simulating a slow / blocking task (configurable via 
// the query param delayMs. 

@Slf4j
@RestController
@RequestMapping("/perf")
public class PerfController {
  private static final String ENV = SpringProfileResolver.environment().asString();
  private static final String FORMULA = SpringProfileResolver.formula().asString();
  private final String name;

  Executor executor;

  public PerfController(
      @Value("${spring.application.name}") String name,
      @Qualifier("completableFutureWorkerPool") Executor executor) {
    this.name = name;
    this.executor = executor;
  }

  @GetMapping("/instant")
  Mono<Bar> perfInstant(
      @RequestParam(name = "delayMs", required = false, defaultValue = "0") Long delayMs,
      ServerHttpRequest serverHttpRequest) {
    log.info(
        "GET "
            + serverHttpRequest.getURI().getRawPath()
            + serverHttpRequest.getURI().getRawQuery());
    return Mono.just(Bar.builder().application(name).environment(ENV).formula(FORMULA).build());
  }

  @GetMapping("/mono-delay")
  Mono<Bar> perfMonoDelay(
      @RequestParam(name = "delayMs", required = false, defaultValue = "0") Long delayMs,
      ServerHttpRequest serverHttpRequest) {
    log.info(
        "GET "
            + serverHttpRequest.getURI().getRawPath()
            + serverHttpRequest.getURI().getRawQuery());
    return Mono.delay(Duration.ofMillis(delayMs))
        .then(Mono.just(Bar.builder().application(name).environment(ENV).formula(FORMULA).build()));
  }

  @GetMapping("/mono-sleep")
  Mono<Bar> perfMonoSleep(
      @RequestParam(name = "delayMs", required = false, defaultValue = "0") Long delayMs,
      ServerHttpRequest serverHttpRequest) {
    log.info(
        "GET "
            + serverHttpRequest.getURI().getRawPath()
            + serverHttpRequest.getURI().getRawQuery());
    return Mono.fromCallable(
        () -> {
          Thread.sleep(delayMs);
          return Bar.builder().application(name).environment(ENV).formula(FORMULA).build();
        });
  }

  @GetMapping("/mono-sleep-sched")
  Mono<Bar> perfMonoSleepShed(
      @RequestParam(name = "delayMs", required = false, defaultValue = "0") Long delayMs,
      ServerHttpRequest serverHttpRequest) {
    log.info(
        "GET "
            + serverHttpRequest.getURI().getRawPath()
            + serverHttpRequest.getURI().getRawQuery());
    return Mono.fromCallable(
            () -> {
              Thread.sleep(delayMs);
              return Bar.builder().application(name).environment(ENV).formula(FORMULA).build();
            })
        .subscribeOn(Schedulers.elastic());
  }

  @GetMapping("/raw-sleep")
  Bar perfSleep(
      @RequestParam(name = "delayMs", required = false, defaultValue = "0") Long delayMs,
      ServerHttpRequest serverHttpRequest)
      throws InterruptedException {
    log.info(
        "GET "
            + serverHttpRequest.getURI().getRawPath()
            + serverHttpRequest.getURI().getRawQuery());
    Thread.sleep(delayMs);
    return Bar.builder().application(name).environment(ENV).formula(FORMULA).build();
  }

  @GetMapping("/cf-sleep")
  CompletableFuture<Bar> perfCF(
      @RequestParam(name = "delayMs", required = false, defaultValue = "0") Long delayMs,
      ServerHttpRequest serverHttpRequest)
      throws InterruptedException {
    log.info(
        "GET "
            + serverHttpRequest.getURI().getRawPath()
            + serverHttpRequest.getURI().getRawQuery());
    return CompletableFuture.supplyAsync(
        () -> {
          try {
            Thread.sleep(delayMs);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          return Bar.builder().application(name).environment(ENV).formula(FORMULA).build();
        });
  }

  @GetMapping("/cf-pooled-sleep")
  CompletableFuture<Bar> perfPooledCF(
      @RequestParam(name = "delayMs", required = false, defaultValue = "0") Long delayMs,
      ServerHttpRequest serverHttpRequest)
      throws InterruptedException {
    log.info(
        "GET "
            + serverHttpRequest.getURI().getRawPath()
            + serverHttpRequest.getURI().getRawQuery());
    return CompletableFuture.supplyAsync(
        () -> {
          try {
            Thread.sleep(delayMs);
          } catch (InterruptedException e) {
            e.printStackTrace();
          }
          return Bar.builder().application(name).environment(ENV).formula(FORMULA).build();
        },
        executor);
  }
}