Play Framework如何限制并发请求数？修改Akka配置不生效怎么解决？

问题根因分析 #

你观测到的256并发上限和你修改的两个配置无关，对应两种不同的场景原因：

• 如果你统计的是服务端接收的上游请求并发数：Akka HTTP服务端默认的max-open-requests参数默认值就是256，这个参数限制的是同时处理的HTTP请求总数，你修改的max-connections仅限制TCP连接数，HTTP/1.1和HTTP/2都支持单连接并发处理多个请求，所以修改后者不生效。
• 如果你统计的是对外调用第三方接口的并发请求数：Play默认的AsyncHttpClient（WSClient底层实现）的全局最大连接数默认值为256，你没有修改WSClient的配置自然不会生效。

对应解决方案 #

方案1：限制服务端接收的上游请求并发数 #

在application.conf中添加以下配置，直接修改请求并发上限：

akka.http.server.max-open-requests = 16 # 替换为你需要的并发数值

如果你的业务都是IO密集型请求，还可以单独给业务逻辑配置独立调度器，避免和其他逻辑抢占资源：

1. 先在application.conf中定义独立调度器：

custom-business-dispatcher {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
    fixed-pool-size = 16 # 和你要的并发数保持一致即可
  }
  throughput = 1
}

2. 注入时指定使用该调度器：

import javax.inject.Named

class MyServiceImpl @Inject()(
  ws: WSClient, 
  @Named("custom-business-dispatcher") implicit val ec: ExecutionContext
) extends MyService with LazyLogging

方案2：限制对外调用第三方接口的并发数 #

如果不需要限制上游请求，仅要控制对外调用的并发，有两种实现方式：

方式A：修改WSClient全局配置

在application.conf中添加配置：

# 全局所有域名的总连接数上限
play.ws.ahc.maxConnectionsTotal = 16
# 单个域名的连接数上限，如果你只调用同一个第三方接口，改这个参数更直接
play.ws.ahc.maxConnectionsPerHost = 16

方式B：用信号量硬控并发

如果不想修改全局配置，也可以用JDK自带的Semaphore做细粒度的并发控制，修改你的代码如下：

import java.util.concurrent.Semaphore

class MyServiceImpl @Inject()(ws: WSClient)(implicit ec: ExecutionContext) extends MyService with LazyLogging {
  // 允许同时最多16个并发请求调用第三方接口
  private val requestSemaphore = new Semaphore(16)
  val atomicInteger = new AtomicInteger(0)

  override def process(request: Request): Response = {
    val requestId = atomicInteger.incrementAndGet()
    logger.info(s"Received request $requestId")
    val futures: Seq[Future[String]] = request.records.map { record =>
      val future = for {
        _ <- Future(requestSemaphore.acquire())
        response <- ws.url("https://www.whatever.com").get()
        result <- response.status match {
          case 200 => Future.successful("successful")
          case other => Future.failed(new RuntimeException(s"Returned response ${other}"))
        }
      } yield result
      // 无论请求成功失败都释放许可，避免死锁
      future.onComplete(_ => requestSemaphore.release())
      future
    }
    // 修正原有统计逻辑的bug：整个请求的所有子任务都完成后再减少计数，避免统计数值为负
    val allTaskDone = Future.sequence(futures)
    allTaskDone.onComplete { _ =>
      val remaining = atomicInteger.decrementAndGet()
      logger.info(s"Processed request ${requestId} ${remaining} remaining")
    }
    Response(allTaskDone)
  }
}

额外说明 #

你原来的代码存在统计逻辑错误：每个请求入口只对atomicInteger加1，但每个子请求成功都会减1，如果一个请求包含N条记录，最终统计的在途请求数会出现负数，上面的代码已经修复了这个问题。

内容的提问来源于stack exchange，提问作者DJ180