SpringWebFluxStress

SpringstressApplication.java

package com.jyuzawa.springstress;

import java.net.URI;
import java.util.Arrays;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.reactive.function.server.RequestPredicates;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.RouterFunctions.Builder;

import org.springframework.web.reactive.function.server.ServerResponse;

import io.netty.channel.DefaultEventLoopGroup;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.util.concurrent.EventExecutor;

// An in-process client sieges the in-process server
@SpringBootApplication
public class SpringstressApplication {

	private static final org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(SpringstressApplication.class);

	private static final HttpHeaders HTTP_HEADERS;
	static {
		HttpHeaders httpHeaders = new HttpHeaders();
		httpHeaders.set("x-foo", "foo");
		httpHeaders.set("x-bar", "bar");
		httpHeaders.set("x-baz", "baz");
		httpHeaders.set("x-boom", "boom");
		HTTP_HEADERS = HttpHeaders.readOnlyHttpHeaders(httpHeaders);
	}

	@Bean
	RouterFunction<ServerResponse> composedRoutes() {
		Builder builder = RouterFunctions.route();
		for (int i = 0; i < 25; i++) {
			builder.add(RouterFunctions.route(
					RequestPredicates.GET("/eijrgijgwpokagpokopkpaorgkopkpargokpokargakagppkga" + i),
					req -> ServerResponse.ok().bodyValue("foo")));
		}
		builder.add(RouterFunctions.route(RequestPredicates.GET("/eijrgijgwpokagpokopkpaorgkopkpargokpokargakagppkgaX"),
				req -> req.bodyToMono(MyPojo.class)
						.flatMap(body -> ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
								.headers(h -> h.addAll(HTTP_HEADERS)).bodyValue(body))));
		for (int i = 25; i < 60; i++) {
			builder.add(RouterFunctions.route(
					RequestPredicates.GET("/eijrgijgwpokagpokopkpaorgkopkpargokpokargakagppkga" + i),
					req -> ServerResponse.ok().bodyValue("bar")));
		}
		return builder.build();
	}

	public static void main(String[] args) throws InterruptedException {
		SpringApplication.run(SpringstressApplication.class, args);
		WebClient webClient = WebClient.builder().defaultHeader("x-blah", "blah").defaultCookie("cookie", "crumbs")
				.build();
		URI uri = URI.create("http://127.0.0.1:8080/eijrgijgwpokagpokopkpaorgkopkpargokpokargakagppkgaX");
		DefaultEventLoopGroup elg = new DefaultEventLoopGroup();
		AtomicInteger count = new AtomicInteger();
		for (EventExecutor el : elg) {
			el.submit(() -> {
				Random r = new Random();
				while (true) {
					MyPojo body = new MyPojo();
					body.foo = 15;
					body.bar = "strgasdfasdgagasdga";
					webClient.method(org.springframework.http.HttpMethod.GET).uri(uri)
							.contentType(MediaType.APPLICATION_JSON).headers(h -> h.putAll(HTTP_HEADERS))
							.bodyValue(body).retrieve().bodyToMono(MyPojo.class).block();
					count.incrementAndGet();
				}
			});
		}
		while (true) {
			Thread.sleep(1000);
			long sum = count.getAndSet(0);
			log.info("qps: {}", sum);
		}
	}

	public static final class MyPojo {

		public int foo;
		public String bar;
	}

}