OkHttp线程池和连接池

QuibblerQuibbler 2021-03-20 约 25 分钟 5215 次阅读

OkHttp线程池和连接池

了解了OkHttp的网络请求流程以及拦截器实现原理,再关注OkHttp中两个重要的:OkHttp的线程池和连接池。

1、OkHttp线程池

在OkHttp网络请求流程一文中,我们分析了OkHttp异步和同步请求流程。请求最后都在Dispatcher中分发调度处理,最后被ExecutorService执行。

1.1、Dispatcher

Dispatcher中执行任务的执行器是executorService。

private var executorServiceOrNull: ExecutorService? = null
    
  val executorService: ExecutorService
    get() {
      if (executorServiceOrNull == null) {
        executorServiceOrNull = ThreadPoolExecutor(0, Int.MAX_VALUE, 60, TimeUnit.SECONDS,
            SynchronousQueue(), threadFactory("$okHttpName Dispatcher", false))
      }
      return executorServiceOrNull!!
    }

Dispatcher有两个构造器:一个默认无参,一个带ExecutorService参数。

constructor()
  
  constructor(executorService: ExecutorService) : this() {
    this.executorServiceOrNull = executorService
  }

1.2、默认线程池

如果没有设置OkHttp线程执行器,其内部有一个默认的executorServiceOrNull,会充当Dispatcher的任务执行器,默认实现如下:

executorServiceOrNull = ThreadPoolExecutor(0, Int.MAX_VALUE, 60, TimeUnit.SECONDS,
            SynchronousQueue(), threadFactory("$okHttpName Dispatcher", false))

无核心线程,最大线程数量无上限,线程空闲存活时间60s,关于线程池详见线程池 Executors。

①核心线程数量0:保持在线程池中的线程数量,为0代表线程空闲后不会保留。

②Integer.MAX_VALUE:表示线程池可以容纳最大线程数量。

③TimeUnit.SECOND:当线程池中的线程数量大于核心线程时,空闲的线程就会等待60s才会被终止,如果小于,则会立刻停止。

④SynchronousQueue:同步队列,按序排队,先来先服务。

⑤threadFactory("$okHttpName Dispatcher", false):线程工厂。

1.3、队列限制

既然默认的executorServiceOrNull线程池无上限,不断的接受网络请求任务会不会导致线程数量太多而引起OOM呢?

不会的,OkHttp会控制进入线程池执行的任务数量,还记得在OkHttp网络请求流程一文中异步请求涉及到的两个异步队列:

/** Ready async calls in the order they'll be run. */
  private val readyAsyncCalls = ArrayDeque<AsyncCall>()
  
  /** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
  private val runningAsyncCalls = ArrayDeque<AsyncCall>()

在promoteAndExecute()方法中,从准备队列readyAsyncCalls取出任务放到执行队列runningAsyncCalls中执行的时候会判断是否达到最大任务数量maxRequests:

private fun promoteAndExecute(): Boolean {
    ...
    val executableCalls = mutableListOf<AsyncCall>()
    val isRunning: Boolean
    synchronized(this) {
      val i = readyAsyncCalls.iterator()
      while (i.hasNext()) {
        val asyncCall = i.next()
        
        // Max capacity.
        if (runningAsyncCalls.size >= this.maxRequests) break 
        
        // Host max capacity.
        if (asyncCall.callsPerHost.get() >= this.maxRequestsPerHost) continue 
        ...
        executableCalls.add(asyncCall)
        runningAsyncCalls.add(asyncCall)
      }
    }
    for (i in 0 until executableCalls.size) {
      val asyncCall = executableCalls[i]
      asyncCall.executeOn(executorService)
    }
    return isRunning
  }

Dispatcher中定了最大任务数量maxRequests,默认64个并发异步任务。同时还限制了对每个主机同时执行的最大请求数maxRequestsPerHost为5,避免服务端承受太大“压力”。 

/**
   * The maximum number of requests to execute concurrently. Above this requests queue in memory,
   * waiting for the running calls to complete.
   *
   * If more than [maxRequests] requests are in flight when this is invoked, those requests will
   * remain in flight.
   */
  @get:Synchronized var maxRequests = 64
    set(maxRequests) {
      require(maxRequests >= 1) { "max < 1: $maxRequests" }
      synchronized(this) {
        field = maxRequests
      }
      promoteAndExecute()
    }
    
  /**
   * The maximum number of requests for each host to execute concurrently. This limits requests by
   * the URL's host name. Note that concurrent requests to a single IP address may still exceed this
   * limit: multiple hostnames may share an IP address or be routed through the same HTTP proxy.
   *
   * If more than [maxRequestsPerHost] requests are in flight when this is invoked, those requests
   * will remain in flight.
   *
   * WebSocket connections to hosts **do not** count against this limit.
   */
  @get:Synchronized var maxRequestsPerHost = 5
    set(maxRequestsPerHost) {
      require(maxRequestsPerHost >= 1) { "max < 1: $maxRequestsPerHost" }
      synchronized(this) {
        field = maxRequestsPerHost
      }
      promoteAndExecute()
    }

为什么没有限制同步任务?因为这得开发者自行使用线程或线程池执行execute(),开发者要合理的使用ThreadPoolExecutor创建应用的线程池。推荐阅读为什么阿里不允许用Executors创建线程池?

1.4、自定义线程池

开发者可以自行在OkHttpClient.Builder配置dispatcher,自定义OkHttp线程池。

val executorService = ThreadPoolExecutor(5, 32, 30, TimeUnit.SECONDS, SynchronousQueue())
    val dispatcher: Dispatcher = Dispatcher(executorService)
    dispatcher.maxRequests = 64
    val okHttpClient = OkHttpClient.Builder()
            .dispatcher(dispatcher)
            .build()

2、OkHttp连接池

再看OkHttp框架中另一个重要的概念:连接池,ConnectionPool。

2.1、连接池的初始化

OkHttp的连接池在OkHttpClient对象创建的时候初始化:

val connectionPool: ConnectionPool = builder.connectionPool

通过OkHttpClient.Builder构造,而OkHttpClient.Builder中使用默认连接池。当然开发者可以自行定义连接池的大小等配置。

class Builder constructor() {
    internal var dispatcher: Dispatcher = Dispatcher()
    internal var connectionPool: ConnectionPool = ConnectionPool()
    ...
  }

调用ConnectionPool默认构造函数:创建一个默认5个连接大小的连接池,每个连接保活5分钟。

class ConnectionPool internal constructor(
    internal val delegate: RealConnectionPool) {
    
    constructor(
      maxIdleConnections: Int,
      keepAliveDuration: Long,
      timeUnit: TimeUnit) : this(RealConnectionPool(
        taskRunner = TaskRunner.INSTANCE,
        maxIdleConnections = maxIdleConnections,
        keepAliveDuration = keepAliveDuration,
        timeUnit = timeUnit))
        
    constructor() : this(5, 5, TimeUnit.MINUTES)
    ...
  }

RealConnectionPool为ConnectionPool的代理类,真正实现一个连接池的功能。使用ConcurrentLinkedQueue作为池保存连接。

/**
   * Holding the lock of the connection being added or removed when mutating this, and check its
   * [RealConnection.noNewExchanges] property. This defends against races where a connection is
   * simultaneously adopted and removed.
   */
  private val connections = ConcurrentLinkedQueue<RealConnection>()

2.2、复用连接池

在OkHttp拦截器Interceptor一文中,梳理了5个框架默认拦截器。OkHttp处理网络流程在哪一步用到了连接池呢?在那篇博客的第2.4节 ConnectInterceptor中。

@Throws(IOException::class)
  override fun intercept(chain: Interceptor.Chain): Response {
    val realChain = chain as RealInterceptorChain
    val exchange = realChain.call.initExchange(chain)
    val connectedChain = realChain.copy(exchange = exchange)
    return connectedChain.proceed(realChain.request)
  }

别看ConnectInterceptor代码就这么点,它可完成了连接池复用、和服务器建立连接等。看一下其中RealCall的initExchange()方法:

/** 查找新的或合并的连接以承载即将到来的请求和响应. */
  internal fun initExchange(chain: RealInterceptorChain): Exchange {
    ...
    val exchangeFinder = this.exchangeFinder!!
    val codec = exchangeFinder.find(client, chain)
    val result = Exchange(this, eventListener, exchangeFinder, codec)
    ...
    return result
  }

内部调用RealCall的ExchangeFinder成员的find()方法,等等!exchangeFinder什么时候初始化的?让我们时光倒流,回到RetryAndFollowUpInterceptor拦截器中:

class RetryAndFollowUpInterceptor(private val client: OkHttpClient) : Interceptor {
    override fun intercept(chain: Interceptor.Chain): Response {
      val realChain = chain as RealInterceptorChain
      var request = chain.request
      val call = realChain.call
      ...
      var newExchangeFinder = true
      ...
      while (true) {
        call.enterNetworkInterceptorExchange(request, newExchangeFinder)
        ...
      }
  }

在该拦截器中调用RealCall的enterNetworkInterceptorExchange(request: Request, newExchangeFinder: Boolean)方法:

fun enterNetworkInterceptorExchange(request: Request, newExchangeFinder: Boolean) {
    ...
    if (newExchangeFinder) {
      this.exchangeFinder = ExchangeFinder(
          connectionPool,
          createAddress(request.url),
          this,
          eventListener
      )
    }
  }

传入的connectionPoll变量,其实就是从OkHttpClient的ConnectionPoll成员中的代理类RealConnectionPool。

private val connectionPool: RealConnectionPool = client.connectionPool.delegate

完成了对RealCall中exchangeFinder成员的初始化,后面的拦截器中可以使用该成员变量。

回到RealCall的initExchange()方法中继续,接着调用开启ExchangeFinder的一系列调用链:find -> findHealthyConnection -> findConnection 。

fun find(client: OkHttpClient,chain: RealInterceptorChain): ExchangeCodec {
    try {
      val resultConnection = findHealthyConnection(
          connectTimeout = chain.connectTimeoutMillis,
          readTimeout = chain.readTimeoutMillis,
          writeTimeout = chain.writeTimeoutMillis,
          pingIntervalMillis = client.pingIntervalMillis,
          connectionRetryEnabled = client.retryOnConnectionFailure,
          doExtensiveHealthChecks = chain.request.method != "GET"
      )
      return resultConnection.newCodec(client, chain)
    }...
  }

从连接池中复用连接,如果没有就新建连接放入池中。这其中的逻辑比较复杂,不需要了解每个实现细节,非硬着头皮钻要花很多时间。

2.3、连接池的维护

最后还有一点,在连接池实现类RealConnectionPool中用Task任务在TaskRunner中执行完成连接池的维护。

private val cleanupTask = object : Task("$okHttpName ConnectionPool") {
    override fun runOnce() = cleanup(System.nanoTime())
  }

还记得2.1节连接池初始化时传入的其中一个参数就是TaskRunner实例:

constructor(
      maxIdleConnections: Int,
      keepAliveDuration: Long,
      timeUnit: TimeUnit) : this(RealConnectionPool(
        taskRunner = TaskRunner.INSTANCE,
        maxIdleConnections = maxIdleConnections,
        keepAliveDuration = keepAliveDuration,
        timeUnit = timeUnit))

TaskRunner的INSTANCE单例使用Kotlin伴生对象实现,这比Java方便多了。

companion object {
    val INSTANCE = TaskRunner(RealBackend(threadFactory("$okHttpName TaskRunner", daemon = true)))
  }

TaskRunner内部持有的Backend接口实现类RealBackend对象,内建了一个执行器和1.2节中说到的OkHttp默认线程池一样。

class RealBackend(threadFactory: ThreadFactory) : Backend {
    private val executor = ThreadPoolExecutor(
        0, // corePoolSize.
        Int.MAX_VALUE, // maximumPoolSize.
        60L, TimeUnit.SECONDS, // keepAliveTime.
        SynchronousQueue(),
        threadFactory
    )
    ...
  }

回到RealConnectionPool中,每当对连接池进行操作,比如添加新的连接放入连接池中,都会执行一遍Task:如果闲置连接超过了保持活动限制或闲置连接限制,则移除最长的连接。

fun put(connection: RealConnection) {
    connection.assertThreadHoldsLock()
    connections.add(connection)
    cleanupQueue.schedule(cleanupTask)
  }

具体执行TaskRunner执行Task的流程就不展开了,也比较有意思。因为这一个TaskRunner,不仅要执行维护连接池的Task、还负责Cache缓存任务的处理,总之在OkHttp中用的地方也不少。

相关推荐

精选
Airbnb开源项目
开源

Airbnb开源项目

Airbnb Engineering & Data ScienceAirflow Use Apache Airflow (incubating) to author workflows as directed acyclic graphs (DAGs) of tasks12,263Airpal Web UI for PrestoDB2,502Aerosolve A machine learning

1.8k
优美的开源动效库:Lottie
开源

优美的开源动效库:Lottie

优美的开源动效库:Lottie1、强大的动效LottieLottie是一个适用于Android,iOS,Web和Windows的库,它可以使用Bodymovin解析以json格式导出的Adobe After Effects动画,并在移动设备和Web上原生渲染它们!GitHub:https://github.com/airbnb/lottie-androidLottie官网:http://airbn

4.3k
Glide配置:AppGlideModule
开源

Glide配置:AppGlideModule

Glide配置:AppGlideModuleGlide用起来非常方便,基本用法见Glide的基本用法,简短的一句链式调用就能给ImageView设置图片。而且自带缓存实现,如果不进行额外的设置Glide的缓存目录、缓存大小都是默认的。应用可以自行进行定制修改,就用到AppGlideModule。1、自定义AppGlideModule配置Glide加载图片的设置、缓存设置、缓存策略,就需要应用自己继

3.8k