With the proliferation of microservices, more and more companies are adopting SpringCloud for their internal microservices framework.
According to the concept of micro-service, the functions of each single application should be divided into micro-services (modules) with independent functions according to the principle of orthogonal functions, that is, the principle of independent functions, and then provide unified external services through interface aggregation.
However, with the increasing number of microservice modules, more and more interfaces need to be aggregated for middle-level services that provide services externally through interface aggregation. Over time, interface aggregation becomes a very difficult performance bottleneck in distributed microservices architectures!
For example, an aggregation Service needs to aggregate the data of Service, Route, and Plugin services to provide external services:
@Headers({ "Accept: application/json" })
public interface ServiceClient {
@RequestLine("GET /")
List<Service> list(a);
}
Copy the code
@Headers({ "Accept: application/json" })
public interface RouteClient {
@RequestLine("GET /")
List<Route> list(a);
}
Copy the code
@Headers({ "Accept: application/json" })
public interface PluginClient {
@RequestLine("GET /")
List<Plugin> list(a);
}
Copy the codeUse declarative OpenFeign instead of HTTP Client for network requests
Writing unit tests
public class SyncFeignClientTest {
public static final String SERVER = "http://devops2:8001";
private ServiceClient serviceClient;
private RouteClient routeClient;
private PluginClient pluginClient;
@Before
public void setup(a){
BasicConfigurator.configure();
Logger.getRootLogger().setLevel(Level.INFO);
String service = SERVER + "/services";
serviceClient = Feign.builder()
.target(ServiceClient.class, service);
String route = SERVER + "/routes";
routeClient = Feign.builder()
.target(RouteClient.class, route);
String plugin = SERVER + "/plugins";
pluginClient = Feign.builder()
.target(PluginClient.class, plugin);
}
@Test
public void aggressionTest(a) {
long current = System.currentTimeMillis();
System.out.println("Start calling aggregate query");
serviceTest();
routeTest();
pluginTest();
System.out.println("End of call to aggregate query! Time:" + (System.currentTimeMillis() - current) + "毫秒");
}
@Test
public void serviceTest(a){
long current = System.currentTimeMillis();
System.out.println("Start getting Service");
String service = serviceClient.list();
System.out.println(service);
System.out.println("Obtaining Service is complete! Time:" + (System.currentTimeMillis() - current) + "毫秒");
}
@Test
public void routeTest(a){
long current = System.currentTimeMillis();
System.out.println("Start getting Route");
String route = routeClient.list();
System.out.println(route);
System.out.println("Obtaining Route completed! Time:" + (System.currentTimeMillis() - current) + "毫秒");
}
@Test
public void pluginTest(a){
long current = System.currentTimeMillis();
System.out.println("Start getting Plugin");
String plugin = pluginClient.list();
System.out.println(plugin);
System.out.println("Plugin access complete! Time:" + (System.currentTimeMillis() - current) + "毫秒"); }}Copy the codeTest results:
Start call aggregate query start get Service {"next":null,"data":[]} Get Service end! Start to fetch Route {"next":null,"data":[]} Get Route end! Start getting Plugin {"next":null,"data":[]} Get Plugin end! Time: 45 ms call aggregate query end! Time: 223 milliseconds Process finished withexit code 0
Copy the codeIt is obvious that the aggregate query takes 223 ms = 134 ms + 44 ms + 45 ms
That is, the request time of the aggregated service is proportional to the number of interfaces, which is obviously unacceptable!
The most common way to solve this problem is to create thread pools in advance and aggregate interfaces through multi-threaded concurrent request interfaces!
This kind of scheme can be found in the Internet casually baidu, today I will not post its code out! But a few downsides to this approach:
The original mainstream Servlet container for JavaWeb uses one thread and one Servlet for each HTTP request! This approach is not too big a problem in the case of low concurrency, but due to the failure of Moore’s Law, the number of threads on a single machine still stays at about 10,000, in today’s web site often tens of millions of clicks, the number of threads on a single machine simply cannot cope with tens of millions of concurrent quantities!
In order to solve the time-consuming problem of interface aggregation, the practice of using thread pool multi-threading concurrent network requests is adding fuel to the fire! Originally only need a thread to get the request, through multi-threaded concurrent interface aggregation, the number of threads required to handle each request to enlarge, rapidly reduce the number of available threads in the system, naturally also reduce the number of concurrent system!
NIO, supported since Java5, and its open source framework, Netty, come to mind! Based on Netty and Reactor models, asynchronous and non-blocking JavaWeb frameworks such as SpringWebFlux appear in the Java ecosystem. Spring5 is also based on SpringWebFlux! With asynchronous non-blocking servers, there is also the asynchronous non-blocking network request client WebClient!
Today I will use WebClient and ReactiveFeign to do an asynchronous non-blocking interface aggregation tutorial:
First, introduce dependencies
<dependency>
<groupId>com.playtika.reactivefeign</groupId>
<artifactId>feign-reactor-core</artifactId>
<version>1.0.30</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.playtika.reactivefeign</groupId>
<artifactId>feign-reactor-webclient</artifactId>
<version>1.0.30</version>
<scope>test</scope>
</dependency>
Copy the codeTo rewrite the Feign client based on Reactor Core, change the List< entity > to FLux< entity >, and change the entity to Mono< entity >.
@Headers({ "Accept: application/json" })
public interface ServiceClient {
@RequestLine("GET /")
Flux<Service> list(a);
}
Copy the code
@Headers({ "Accept: application/json" })
public interface RouteClient {
@RequestLine("GET /")
Flux<Service> list(a);
}
Copy the code
@Headers({ "Accept: application/json" })
public interface PluginClient {
@RequestLine("GET /")
Flux<Service> list(a);
}
Copy the codeThen write the unit tests
public class AsyncFeignClientTest {
public static final String SERVER = "http://devops2:8001";
private CountDownLatch latch;
private ServiceClient serviceClient;
private RouteClient routeClient;
private PluginClient pluginClient;
@Before
public void setup(a){
BasicConfigurator.configure();
Logger.getRootLogger().setLevel(Level.INFO);
latch= new CountDownLatch(3);
String service= SERVER + "/services";
serviceClient= WebReactiveFeign
.<ServiceClient>builder()
.target(ServiceClient.class, service);
String route= SERVER + "/routes";
routeClient= WebReactiveFeign
.<RouteClient>builder()
.target(RouteClient.class, route);
String plugin= SERVER + "/plugins";
pluginClient= WebReactiveFeign
.<PluginClient>builder()
.target(PluginClient.class, plugin);
}
@Test
public void aggressionTest(a) throws InterruptedException {
long current= System.currentTimeMillis();
System.out.println("Start calling aggregate query");
serviceTest();
routeTest();
pluginTest();
latch.await();
System.out.println("End of call to aggregate query! Time:" + (System.currentTimeMillis() - current) + "毫秒");
}
@Test
public void serviceTest(a){
long current= System.currentTimeMillis();
System.out.println("Start getting Service");
serviceClient.list()
.subscribe(result ->{
System.out.println(result);
latch.countDown();
System.out.println("Obtaining Service is complete! Time:" + (System.currentTimeMillis() - current) + "毫秒");
});
}
@Test
public void routeTest(a){
long current= System.currentTimeMillis();
System.out.println("Start getting Route");
routeClient.list()
.subscribe(result ->{
System.out.println(result);
latch.countDown();
System.out.println("Obtaining Route completed! Time:" + (System.currentTimeMillis() - current) + "毫秒");
});
}
@Test
public void pluginTest(a){
long current= System.currentTimeMillis();
System.out.println("Start getting Plugin");
pluginClient.list()
.subscribe(result ->{
System.out.println(result);
latch.countDown();
System.out.println("Plugin access complete! Time:" + (System.currentTimeMillis() - current) + "毫秒"); }); }}Copy the codeThe key point here is that the request that was blocked synchronously is now asynchronous and non-blocked, so you need to use CountDownLatch to synchronize, and call countdownlatch.coutdown () when you get the interface. Call countdownlatch.await () after invoking all interface requests and wait for all interfaces to return results before doing the next step!
Test results:
Plugin {start call aggregate query start fetch Service start fetch Route start fetch Plugin {"next":null,"data": []} {"next":null,"data":[]} Get Plugin end! Time: 215 ms {"next":null,"data":[]} Get Route end! Time: 216 ms Obtaining Service End! Time: 1000 ms call aggregate query end! Time: 1000 ms Process Finished withexit code 0
Copy the codeClearly, the aggregate query time is no longer equal to the sum of all the interface requests, but the maximum of the interface request time!
Let’s start the performance test:
Normal Feign interface aggregation test calls 1000 times:
Start call aggregate query start get Service {"next":null,"data":[]} Get Service end! Start Route {"next":null,"data":[]} Get Route end! Start fetching Plugin {"next":null,"data":[]} Get Plugin end! Time: 93 ms Call aggregate query end! Time: 343 ms Summary: 238515, Average: 238Copy the codeQuery interface aggregation 1000 times using WebClient:
Plugin {start call aggregate query start fetch Service start fetch Route start fetch Plugin {"next":null,"data": []} {"next":null,"data":[]} Get Route end! Time: 122 ms {"next":null,"data":[]} Get Service end! Time: 122 ms To obtain Plugin end! Time: 121 ms call aggregate query end! Time: 123 ms Summary: 89081, Average: 89Copy the codeIn the test results, The WebClient test results were exactly one third as good as the regular FeignClient! Just as expected!