当前位置：首页 > article >正文

[java] 简单的熔断器scala语言案例

article 2024/12/21 9:38:30

failureRateInterval时间内如果addEx(错误)达到 maxFailuresPerInterval 次数，则fused方法返回true,表示触发熔断，进入冷却期coolingInterval，冷却期内fused方法返回true，冷却期过后进入下一个错误统计周期。

scala语言完成

import scala.collection.mutable
case class Excp(ts: Long, throwable: Throwable)

trait ExceptionStrategy {
}

case class FailureRateExit(maxFailuresPerInterval: Int, failureRateInterval: Int, coolingInterval: Int) extends ExceptionStrategy {
  val LOG: Logger = org.slf4j.LoggerFactory.getLogger(this.getClass)


  private val exceptionQueue: mutable.Queue[Excp] = mutable.Queue.empty
  // exceptionQueue += Excp(0L, new RuntimeException())
  private var lastFuseTs: Long = 0L

  def fused(): Boolean = {
    val curr = System.currentTimeMillis()
    if ((curr - lastFuseTs) / 1000 <= coolingInterval) {
      LOG.info(s"fusing cooling, fist error ts:${exceptionQueue.head.ts}, last error ts:${exceptionQueue.last.ts}")
      exceptionQueue.clear()
      return true
    }
    cleanup(curr)
    val trigger = exceptionQueue.size >= maxFailuresPerInterval && exceptionQueue.head.ts - exceptionQueue.last.ts <= failureRateInterval
    if (trigger) {
      lastFuseTs = curr
      LOG.info(s"fusing triggered, fist error ts:${exceptionQueue.head.ts}, last error ts:${exceptionQueue.last.ts}, total errors:${exceptionQueue.size}")
    }
    trigger
  }

  def addEx(ex: Exception): Unit = {
    val curr = System.currentTimeMillis()
    exceptionQueue += Excp(curr, ex)
    cleanup(curr)
  }

  private def cleanup(curr: Long = System.currentTimeMillis()): Unit = {
    exceptionQueue.dequeueAll(e => (curr - e.ts) / 1000 >= failureRateInterval)
    exceptionQueue.dequeueAll(e => exceptionQueue.size > maxFailuresPerInterval * 2 && e != exceptionQueue.last && e != exceptionQueue.head)
  }
}

测试代码：

import org.scalatest.funsuite.AnyFunSuite
import scala.util.Random

class StrategyTest extends AnyFunSuite {

  test("fail-rate") {
    val f = FailureRateExit(3, 10, 15)
    while (true) {
      f.addEx(new RuntimeException())
      println(f.fused())
      Thread.sleep(Random.nextInt(8 * 1000))
    }
  }
}

测试结果：

false
false
false
false
13:52:03.519 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:30 --- fusing triggered, fist error ts:1734414718275, last error ts:1734414723512, total errors:3
true
13:52:10.202 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414722743, last error ts:1734414730202
true
13:52:18.120 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414738120, last error ts:1734414738120
true
false
false
false
false
false
false
13:52:55.780 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:30 --- fusing triggered, fist error ts:1734414768116, last error ts:1734414775780, total errors:3
true
13:53:00.976 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414775249, last error ts:1734414780976
true
13:53:03.931 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414783931, last error ts:1734414783931
true
13:53:07.628 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414787628, last error ts:1734414787628
true
false
false
false
13:53:28.360 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:30 --- fusing triggered, fist error ts:1734414800129, last error ts:1734414808359, total errors:3
true
13:53:35.734 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414808359, last error ts:1734414815734
true
13:53:42.517 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414822514, last error ts:1734414822514
true
false
false
13:53:49.223 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:30 --- fusing triggered, fist error ts:1734414824804, last error ts:1734414829223, total errors:3
true
13:53:50.351 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414824804, last error ts:1734414830351
true
13:53:57.470 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414837470, last error ts:1734414837470
true
13:54:04.618 [ScalaTest-run-running-StrategyTest] [INFO ] FailureRateExit:22 --- fusing cooling, fist error ts:1734414844618, last error ts:1734414844618
true
false
false

使用方式伪代码：

  private val strategy: FailureRateExit = FailureRateExit(failureRateInterval, maxFailuresPerInterval, failCoolingInterval)
       
	   try {
          if(!strategy.fused){
		  // 执行正常逻辑
		  }else{
		  // 忽略
		  }
        } catch {
          case ex: Exception =>
		    // 添加异常到熔断器
            strategy.addEx(ex)
        }