spring cloud resilience4j - Bulkhead 线程隔离并发控制

末蓝、 2022-12-28 11:18 199阅读 0赞

# 简介 #

resilience4j 的 Bulkhead 模块实现并发控制，用于限制方法调用的并发数。

Bulkhead隔离不同种类的调用，并进行流控，这样可以避免某类调用异常或占用过多资源，危及系统整体。

实现方式用两种：

*  通过信号量Semaphores控制
 *  通过线程池控制，使用一个有界队列和一个固定数量线程池。

# 信号量与线程池比较 #

## 信号量模式 ##

信号量Semaphore是一个并发工具类，用来控制可同时并发的线程数，其内部维护了一组虚拟许可，通过构造器指定许可的数量，每次线程执行操作时先通过acquire方法获得许可，执行完毕再通过release方法释放许可。如果无可用许可，那么acquire方法将一直阻塞，直到其它线程释放许可。

在该模式下，接收请求和执行下游依赖在同一个线程内完成，不存在线程上下文切换所带来的性能开销，所以大部分场景应该选择信号量模式。

由于该模式不创建线程，不适合并行任务的情况。

## 线程池模式 ##

线程池用来控制实际工作的线程数量，通过线程复用的方式来减小内存开销。线程池可同时工作的线程数量是一定的，超过该数量的线程需进入线程队列等待，直到有可用的工作线程来执行任务。

该模式的优点是支持异步任务，缺点是线程调度会有额外的开销。

# 配置参数说明 #

## 信号量 ##

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>maxConcurrentCalls</td> 
   <td>25</td> 
   <td>最大并发</td> 
  </tr> 
  <tr> 
   <td>maxWaitDuration</td> 
   <td>0</td> 
   <td>获取信号量的最长排队等待时间</td> 
  </tr> 
 </tbody> 
</table>

## 线程池 ##

<table> 
 <thead> 
  <tr> 
   <th>属性</th> 
   <th>默认值</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>maxThreadPoolSize</td> 
   <td>Runtime.getRuntime().availableProcessors()</td> 
   <td>最大线程数</td> 
  </tr> 
  <tr> 
   <td>coreThreadPoolSize</td> 
   <td>Runtime.getRuntime().availableProcessors() - 1</td> 
   <td>最少线程数</td> 
  </tr> 
  <tr> 
   <td>queueCapacity</td> 
   <td>100</td> 
   <td>等待队列大小</td> 
  </tr> 
  <tr> 
   <td>keepAliveDuration</td> 
   <td>20[ms]</td> 
   <td>线程空闲时，销毁前的等待时间</td> 
  </tr> 
 </tbody> 
</table>

# 实例 #

## 依赖 ##

spring boot:

repositories { 
        jcenter()
    }
    
    dependencies { 
      compile "io.github.resilience4j:resilience4j-spring-boot2:${resilience4jVersion}"
      compile('org.springframework.boot:spring-boot-starter-actuator')
      compile('org.springframework.boot:spring-boot-starter-aop')
    }

spring cloud:

repositories { 
        jCenter()
    }
    
    dependencies { 
        compile "io.github.resilience4j:resilience4j-spring-cloud2:${resilience4jVersion}"
        compile('org.springframework.boot:spring-boot-starter-actuator')
        compile('org.springframework.boot:spring-boot-starter-aop')
        compile('org.springframework.cloud:spring-cloud-starter-config')  
    }

## 配置文件 ##

resilience4j.bulkhead:
        instances:
            backendA:
                maxConcurrentCalls: 10
            backendB:
                maxWaitDuration: 10ms
                maxConcurrentCalls: 20
                
    resilience4j.thread-pool-bulkhead:
      instances:
        backendC:
          maxThreadPoolSize: 1
          coreThreadPoolSize: 1
          queueCapacity: 1

## 代码 ##

@Bulkhead(name = "backendA",fallbackMethod = "fallback")
    public String method(String param1) { 
        return "test";
    }
    
    private String fallback(String param1, IllegalArgumentException e) { 
        return "test fallback";
    }
    
    private String fallback(String param1, Exception e) { 
        return "test fallback";
    }

上面配置了`method`方法只允许10个并发。如果超过10个并发，会进入第二个fallback方法；如果方法抛出`IllegalArgumentException`,则会进入第一个fallback方法。

# 更多 #

[spring cloud resilience4j-retry 重试][spring cloud resilience4j-retry]  
[spring cloud resilience4j - Bulkhead 线程隔离 并发控制][spring cloud resilience4j - Bulkhead _]  
[spring cloud Resilience4j - 熔断器 CircuitBreaker][spring cloud Resilience4j - _ CircuitBreaker]  
[spring cloud resilience4j - RateLimiter 流控][spring cloud resilience4j - RateLimiter]

https://resilience4j.readme.io/docs/bulkhead

[spring cloud resilience4j-retry]: https://blog.csdn.net/xgw1010/article/details/111464801
[spring cloud resilience4j - Bulkhead _]: https://blog.csdn.net/xgw1010/article/details/111300196
[spring cloud Resilience4j - _ CircuitBreaker]: https://blog.csdn.net/xgw1010/article/details/108675113
[spring cloud resilience4j - RateLimiter]: https://blog.csdn.net/xgw1010/article/details/111590574