专栏目录
20. 启动一个 Eureka Server 集群 26.OpenFeign的组件 32. 改进负载均衡算法 27.OpenFeign的生命周期-创建代理 33. 实现重试、断路器以及线程隔离源码 8.理解 NamedContextFactory 25.OpenFeign简介与使用 14.UnderTow AccessLog 配置介绍 17.Eureka的实例配置 1. 背景 21.Spring Cloud LoadBalancer简介 37. 实现异步的客户端封装配置管理的意义与设计 44.避免链路信息丢失做的设计(2) 6.微服务特性相关的依赖说明 10.使用Log4j2以及一些核心配置 40. spock 单元测试封装的 WebClient(上) 11.Log4j2 监控相关 3.Eureka Server 与 API 网关要考虑的问题 15.UnderTow 订制 45. 实现公共日志记录 16.Eureka架构和核心概念 9.如何理解并定制一个Spring Cloud组件 12.UnderTow 简介与内部原理 38. 实现自定义 WebClient 的 NamedContextFactory 39. 改造 resilience4j 粘合 WebClient 19.Eureka的服务端设计与配置 41. SpringCloudGateway 基本流程讲解(2) 29.Spring Cloud OpenFeign 的解析(1) 42.SpringCloudGateway 现有的可供分析的请求日志以及缺陷 35. 验证线程隔离正确性 23.订制Spring Cloud LoadBalancer 40. spock 单元测试封装的 WebClient(下) 28.OpenFeign的生命周期-进行调用 7.从Bean到SpringCloud 44.避免链路信息丢失做的设计(1) 41. SpringCloudGateway 基本流程讲解(1) 24.测试Spring Cloud LoadBalancer 36. 验证断路器正确性 30. FeignClient 实现重试 5.所有项目的parent与spring-framework-common说明 22.Spring Cloud LoadBalancer核心源码 4.maven依赖回顾以及项目框架结构 18.Eureka的客户端核心设计和配置 13.UnderTow 核心配置 43.为何 SpringCloudGateway 中会有链路信息丢失 2.微服务框架需要考虑的问题 34.验证重试配置正确性 31. FeignClient 实现断路器以及线程隔离限流的思路

35. 验证线程隔离正确性

unknown
• 阅读 1002

35. 验证线程隔离正确性

本系列代码地址:https://github.com/JoJoTec/spring-cloud-parent

上一节我们通过单元测试验证了重试的正确性,这一节我们来验证我们线程隔离的正确性,主要包括:

  1. 验证配置正确加载:即我们在 Spring 配置(例如 application.yml)中的加入的 Resilience4j 的配置被正确加载应用了。
  2. 相同微服务调用不同实例的时候,使用的是不同的线程(池)。

验证配置正确加载

与之前验证重试类似,我们可以定义不同的 FeignClient,之后检查 resilience4j 加载的线程隔离配置来验证线程隔离配置的正确加载。

并且,与重试配置不同的是,通过系列前面的源码分析,我们知道 spring-cloud-openfeign 的 FeignClient 其实是懒加载的。所以我们实现的线程隔离也是懒加载的,需要先调用,之后才会初始化线程池。所以这里我们需要先进行调用之后,再验证线程池配置。

首先定义两个 FeignClient,微服务分别是 testService1 和 testService2,contextId 分别是 testService1Client 和 testService2Client

@FeignClient(name = "testService1", contextId = "testService1Client")
public interface TestService1Client {
    @GetMapping("/anything")
    HttpBinAnythingResponse anything();
}
@FeignClient(name = "testService2", contextId = "testService2Client")
    public interface TestService2Client {
        @GetMapping("/anything")
        HttpBinAnythingResponse anything();
}

然后,我们增加 Spring 配置,并且给两个微服务都添加一个实例,使用 SpringExtension 编写单元测试类:

//SpringExtension也包含了 Mockito 相关的 Extension,所以 @Mock 等注解也生效了
@ExtendWith(SpringExtension.class)
@SpringBootTest(properties = {
        //默认请求重试次数为 3
        "resilience4j.retry.configs.default.maxAttempts=3",
        // testService2Client 里面的所有方法请求重试次数为 2
        "resilience4j.retry.configs.testService2Client.maxAttempts=2",
        //默认线程池配置
        "resilience4j.thread-pool-bulkhead.configs.default.coreThreadPoolSize=10",
        "resilience4j.thread-pool-bulkhead.configs.default.maxThreadPoolSize=10",
        "resilience4j.thread-pool-bulkhead.configs.default.queueCapacity=1" ,
        //testService2Client 的线程池配置
        "resilience4j.thread-pool-bulkhead.configs.testService2Client.coreThreadPoolSize=5",
        "resilience4j.thread-pool-bulkhead.configs.testService2Client.maxThreadPoolSize=5",
        "resilience4j.thread-pool-bulkhead.configs.testService2Client.queueCapacity=1",
})
@Log4j2
public class OpenFeignClientTest {
    @SpringBootApplication
    @Configuration
    public static class App {
        @Bean
        public DiscoveryClient discoveryClient() {
            //模拟两个服务实例
            ServiceInstance service1Instance1 = Mockito.spy(ServiceInstance.class);
            ServiceInstance service2Instance2 = Mockito.spy(ServiceInstance.class);
            Map<String, String> zone1 = Map.ofEntries(
                    Map.entry("zone", "zone1")
            );
            when(service1Instance1.getMetadata()).thenReturn(zone1);
            when(service1Instance1.getInstanceId()).thenReturn("service1Instance1");
            when(service1Instance1.getHost()).thenReturn("www.httpbin.org");
            when(service1Instance1.getPort()).thenReturn(80);
            when(service2Instance2.getInstanceId()).thenReturn("service1Instance2");
            when(service2Instance2.getHost()).thenReturn("httpbin.org");
            when(service2Instance2.getPort()).thenReturn(80);
            DiscoveryClient spy = Mockito.spy(DiscoveryClient.class);
            Mockito.when(spy.getInstances("testService1"))
                    .thenReturn(List.of(service1Instance1));
            Mockito.when(spy.getInstances("testService2"))
                    .thenReturn(List.of(service2Instance2));
            return spy;
        }
    }
}

编写测试代码,验证配置正确:

@Test
public void testConfigureThreadPool() {
    //防止断路器影响
    circuitBreakerRegistry.getAllCircuitBreakers().asJava().forEach(CircuitBreaker::reset);
    //调用下这两个 FeignClient 确保对应的 NamedContext 被初始化
    testService1Client.anything();
    testService2Client.anything();
    //验证线程隔离的实际配置,符合我们的填入的配置
    ThreadPoolBulkhead threadPoolBulkhead = threadPoolBulkheadRegistry.getAllBulkheads().asJava()
            .stream().filter(t -> t.getName().contains("service1Instance1")).findFirst().get();
    Assertions.assertEquals(threadPoolBulkhead.getBulkheadConfig().getCoreThreadPoolSize(), 10);
    Assertions.assertEquals(threadPoolBulkhead.getBulkheadConfig().getMaxThreadPoolSize(), 10);
    threadPoolBulkhead = threadPoolBulkheadRegistry.getAllBulkheads().asJava()
            .stream().filter(t -> t.getName().contains("service1Instance2")).findFirst().get();
    Assertions.assertEquals(threadPoolBulkhead.getBulkheadConfig().getCoreThreadPoolSize(), 5);
    Assertions.assertEquals(threadPoolBulkhead.getBulkheadConfig().getMaxThreadPoolSize(), 5);
}

相同微服务调用不同实例的时候,使用的是不同的线程(池)。

我们需要确保,最后调用(也就是发送 http 请求)的执行的线程池,必须是对应的 ThreadPoolBulkHead 中的线程池。这个需要我们对 ApacheHttpClient 做切面实现,添加注解 @EnableAspectJAutoProxy(proxyTargetClass = true)

//SpringExtension也包含了 Mockito 相关的 Extension,所以 @Mock 等注解也生效了
@ExtendWith(SpringExtension.class)
@SpringBootTest(properties = {
        //默认请求重试次数为 3
        "resilience4j.retry.configs.default.maxAttempts=3",
        // testService2Client 里面的所有方法请求重试次数为 2
        "resilience4j.retry.configs.testService2Client.maxAttempts=2",
        //默认线程池配置
        "resilience4j.thread-pool-bulkhead.configs.default.coreThreadPoolSize=10",
        "resilience4j.thread-pool-bulkhead.configs.default.maxThreadPoolSize=10",
        "resilience4j.thread-pool-bulkhead.configs.default.queueCapacity=1" ,
        //testService2Client 的线程池配置
        "resilience4j.thread-pool-bulkhead.configs.testService2Client.coreThreadPoolSize=5",
        "resilience4j.thread-pool-bulkhead.configs.testService2Client.maxThreadPoolSize=5",
        "resilience4j.thread-pool-bulkhead.configs.testService2Client.queueCapacity=1",
})
@Log4j2
public class OpenFeignClientTest {
    @SpringBootApplication
    @Configuration
    @EnableAspectJAutoProxy(proxyTargetClass = true)
    public static class App {
        @Bean
        public DiscoveryClient discoveryClient() {
            //模拟两个服务实例
            ServiceInstance service1Instance1 = Mockito.spy(ServiceInstance.class);
            ServiceInstance service2Instance2 = Mockito.spy(ServiceInstance.class);
            Map<String, String> zone1 = Map.ofEntries(
                    Map.entry("zone", "zone1")
            );
            when(service1Instance1.getMetadata()).thenReturn(zone1);
            when(service1Instance1.getInstanceId()).thenReturn("service1Instance1");
            when(service1Instance1.getHost()).thenReturn("www.httpbin.org");
            when(service1Instance1.getPort()).thenReturn(80);
            when(service2Instance2.getInstanceId()).thenReturn("service1Instance2");
            when(service2Instance2.getHost()).thenReturn("httpbin.org");
            when(service2Instance2.getPort()).thenReturn(80);
            DiscoveryClient spy = Mockito.spy(DiscoveryClient.class);
            Mockito.when(spy.getInstances("testService1"))
                    .thenReturn(List.of(service1Instance1));
            Mockito.when(spy.getInstances("testService2"))
                    .thenReturn(List.of(service2Instance2));
            return spy;
        }
    }
}

拦截 ApacheHttpClientexecute 方法,这样可以拿到真正负责 http 调用的线程池,将线程其放入请求的 Header:

@Aspect
public static class ApacheHttpClientAop {
    //在最后一步 ApacheHttpClient 切面
    @Pointcut("execution(* com.github.jojotech.spring.cloud.webmvc.feign.ApacheHttpClient.execute(..))")
    public void annotationPointcut() {
    }

    @Around("annotationPointcut()")
    public Object around(ProceedingJoinPoint pjp) throws Throwable {
        //设置 Header,不能通过 Feign 的 RequestInterceptor,因为我们要拿到最后调用 ApacheHttpClient 的线程上下文
        Request request = (Request) pjp.getArgs()[0];
        Field headers = ReflectionUtils.findField(Request.class, "headers");
        ReflectionUtils.makeAccessible(headers);
        Map<String, Collection<String>> map = (Map<String, Collection<String>>) ReflectionUtils.getField(headers, request);
        HashMap<String, Collection<String>> stringCollectionHashMap = new HashMap<>(map);
        stringCollectionHashMap.put(THREAD_ID_HEADER, List.of(String.valueOf(Thread.currentThread().getName())));
        ReflectionUtils.setField(headers, request, stringCollectionHashMap);
        return pjp.proceed();
    }
}

这样,我们就能拿到具体承载请求的线程的名称,从名称中可以看出他所处于的线程池(格式为“bulkhead-线程隔离名称-n”,例如 bulkhead-testService1Client:www.httpbin.org:80-1),接下来我们就来看下不同的实例是否用了不同的线程池进行调用:

@Test
public void testDifferentThreadPoolForDifferentInstance() throws InterruptedException {
    //防止断路器影响
    circuitBreakerRegistry.getAllCircuitBreakers().asJava().forEach(CircuitBreaker::reset);
    Set<String> threadIds = Sets.newConcurrentHashSet();
    Thread[] threads = new Thread[100];
    //循环100次
    for (int i = 0; i < 100; i++) {
        threads[i] = new Thread(() -> {
            Span span = tracer.nextSpan();
            try (Tracer.SpanInScope cleared = tracer.withSpanInScope(span)) {
                HttpBinAnythingResponse response = testService1Client.anything();
                //因为 anything 会返回我们发送的请求实体的所有内容,所以我们能获取到请求的线程名称 header
                String threadId = response.getHeaders().get(THREAD_ID_HEADER);
                threadIds.add(threadId);
            }
        });
        threads[i].start();
    }
    for (int i = 0; i < 100; i++) {
        threads[i].join();
    }
    //确认实例 testService1Client:httpbin.org:80 线程池的线程存在
    Assertions.assertTrue(threadIds.stream().anyMatch(s -> s.contains("testService1Client:httpbin.org:80")));
    //确认实例 testService1Client:httpbin.org:80 线程池的线程存在
    Assertions.assertTrue(threadIds.stream().anyMatch(s -> s.contains("testService1Client:www.httpbin.org:80")));
}

这样,我们就成功验证了,实例调用的线程池隔离。

微信搜索“我的编程喵”关注公众号,每日一刷,轻松提升技术,斩获各种offer

35. 验证线程隔离正确性

点赞
收藏
评论区
推荐文章

暂无数据

unknown
unknown
Lv1
男 · rrrr · rrrrrrrr
rrrrr
文章
0
粉丝
17
获赞
0
热门文章

暂无数据