能否让Retrofit忽略OkHttp Dispatcher？协程请求并行控制问询

1. 调整OkHttpClient配置，移除自身线程池依赖 #

OkHttp默认会用自带的Dispatcher线程池执行请求，这会绕过协程的并行度限制。我们需要修改OkHttpClient配置，让它不再切换到自身线程池，而是直接在调用线程执行请求：

val okHttpClient = OkHttpClient.Builder()
    .dispatcher(Dispatcher().apply {
        // 取消OkHttp自身的并发限制，交由协程控制
        maxRequests = Int.MAX_VALUE
        maxRequestsPerHost = Int.MAX_VALUE
        // 自定义执行器，直接在调用线程执行任务，不走OkHttp线程池
        executorService = object : Executor {
            override fun execute(command: Runnable) {
                command.run()
            }
        }
    })
    .build()

2. 适配Retrofit调用逻辑，关联协程取消与请求取消 #

将Retrofit接口改为返回Call<T>（而非suspend函数），然后在协程中同步执行请求，并通过协程的取消回调关联OkHttp请求的取消操作：

// Retrofit接口定义
interface RetrofitWebserviceApi {
    @GET("entries")
    fun getEntries(@Query("description") description: String): Call<EntriesResponse>

    @GET("categories")
    fun getCategories(): Call<CategoriesResponse>
}

// 封装的NetworkApi类
class NetworkApi(
    private val retrofitWebserviceApi: RetrofitWebserviceApi,
    threadPoolExecutor: ThreadPoolExecutor,
    private val dispatcher: CoroutineDispatcher = threadPoolExecutor.asCoroutineDispatcher()
        .limitedParallelism(CoroutineDispatcherConfig.ioDispatcherLimit),
    // 限制getEntries仅1个并行
    private val singleEntryDispatcher: CoroutineDispatcher = dispatcher.limitedParallelism(1),
) {
    suspend fun getEntries(description: String): EntriesResponse = withContext(singleEntryDispatcher) {
        val call = retrofitWebserviceApi.getEntries(description)
        // 协程取消时同步取消OkHttp请求
        invokeOnCancellation {
            call.cancel()
        }
        call.execute().body() ?: throw IOException("Empty response body")
    }

    suspend fun getCategories(): CategoriesResponse = withContext(dispatcher) {
        val call = retrofitWebserviceApi.getCategories()
        invokeOnCancellation {
            call.cancel()
        }
        call.execute().body() ?: throw IOException("Empty response body")
    }
}

3. 核心原理说明 #

• 自定义OkHttp的Executor后，请求会直接在协程Dispatcher分配的线程上执行，不再依赖OkHttp自身的线程池，这样协程的limitedParallelism()就能精准控制并发数。
• invokeOnCancellation会在协程被取消时触发，此时调用call.cancel()可以同步终止OkHttp的网络请求，完全保留请求取消能力。
• 不同端点使用不同的协程Dispatcher，比如getEntries用单并行Dispatcher，getCategories用全局IO池，确保大量getEntries请求不会阻塞其他接口调用。

可选优化：兼容现有suspend接口 #

如果不想修改已有的Retrofit suspend接口，可以自定义CallAdapter.Factory，让suspend函数内部自动使用同步执行逻辑，无需改动接口定义：

class SyncCallAdapterFactory : CallAdapter.Factory() {
    override fun get(
        returnType: Type,
        annotations: Array<Annotation>,
        retrofit: Retrofit
    ): CallAdapter<*, *>? {
        if (getRawType(returnType) != Call::class.java) return null
        val responseType = getParameterUpperBound(0, returnType as ParameterizedType)
        return object : CallAdapter<Any, Call<Any>> {
            override fun responseType() = responseType
            override fun adapt(call: Call<Any>): Call<Any> = call
        }
    }
}

添加到Retrofit构建器中即可：

val retrofit = Retrofit.Builder()
    .baseUrl(BASE_URL)
    .client(okHttpClient)
    .addCallAdapterFactory(SyncCallAdapterFactory())
    .addConverterFactory(GsonConverterFactory.create())
    .build()

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