Android中Ktor在Worker内发起HTTP请求超时问题排查

我开发的Android应用采用Ktor框架发起HTTP请求，在Worker组件外调用createBucket接口时运行正常，但在Worker内调用uploadImages和downloadImages接口时，始终出现连接超时错误：

io.ktor.client.network.sockets.ConnectTimeoutException: Connect timeout has expired [url=https://api.backblazeb2.com/b2api/v3/b2_authorize_account, connect_timeout=unknown ms]

所有接口均会调用getClient方法，超时问题发生在getClient函数内的请求环节，请求协助排查。

private class B2Client(
    val authorizationToken: String,
    val apiUrl: String,
    val accountId: String,
    val client: HttpClient
)


class BackblazeB2(val remote: Remote) :
    CloudProvider {
    private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())

    override suspend fun getRemoteImagesIds(): List<UUID> {
        val url: String = "/b2api/v3/b2_list_file_names"

        val response = scope.async(Dispatchers.IO) {
            val b2Client = getClient(scope, remote.apiKeyId, remote.apiKey).await()

            return@async b2Client.client.post(b2Client.apiUrl + url) {
                headers {
                    append(HttpHeaders.Authorization, b2Client.authorizationToken)
                }
                contentType(ContentType.Application.Json)
                setBody(
                    buildJsonObject {
                        put("bucketId", remote.bucketId)
                    }
                )
            }
        }.await()

        val body: JsonElement = response.body()

        val files = body.jsonObject.getValue("files").jsonArray

        val imagesIds = mutableListOf<UUID>()

        files.forEach { file ->
            val fileName: String = file.jsonObject.getValue("fileName").jsonPrimitive.content

            imagesIds.add(UUID.fromString(fileName))
        }

        return imagesIds
    }

    override suspend fun uploadImages(images: List<Image>, context: Context): List<Deferred<UUID>> {
        val uploadUrlUrl = "/b2api/v3/b2_get_upload_url"

        val jobs: MutableList<Deferred<UUID>> = mutableListOf()

        images.forEach { image ->
            jobs.add(scope.async(Dispatchers.IO) {
                val b2Client = getClient(scope, remote.apiKeyId, remote.apiKey).await()

                val response = b2Client.client.get(b2Client.apiUrl + uploadUrlUrl) {
                    headers {
                        append(HttpHeaders.Authorization, b2Client.authorizationToken)
                    }
                    url {
                        parameters.append("bucketId", remote.bucketId)
                    }
                }

                val body: JsonElement = response.body()

                val uploadUrl: String = body.jsonObject.getValue("uploadUrl").jsonPrimitive.content
                val authorizationToken: String =
                    body.jsonObject.getValue("authorizationToken").jsonPrimitive.content

                val imageBytes = getImageBytes(image.localPath!!, context)
                val imageSHA1 = calculateSHA1(imageBytes)

                val uploadResponse = b2Client.client.post(uploadUrl) {
                    headers {
                        append(HttpHeaders.Authorization, authorizationToken)
                        append(HttpHeaders.ContentType, image.contentType)
                        append("X-Bz-File-Name", image.uuid.toString())
                        append(HttpHeaders.ContentLength, imageBytes.size.toString())
                        append("X-Bz-Content-Sha1", sha1ToHex(imageSHA1!!))
                    }
                    setBody(imageBytes)
                }

                if (uploadResponse.status != HttpStatusCode.OK) {
                    throw Error("Image upload failed")
                }

                return@async image.uuid
            })
        }

        return jobs
    }

    override suspend fun downloadImages(
        imagesIds: List<UUID>,
        context: Context
    ): List<Deferred<Image>> {
        val downloadUrl = "/file"

        val images: MutableList<Deferred<Image>> = mutableListOf()

        imagesIds.forEach { imageId ->
            images.add(scope.async(Dispatchers.IO) {
                val b2Client = getClient(scope, remote.apiKeyId, remote.apiKey).await()

                val downloadResponse = b2Client.client.get(b2Client.apiUrl + downloadUrl) {
                    url {
                        appendPathSegments(remote.name, imageId.toString())
                    }
                    headers {
                        append(HttpHeaders.Authorization, b2Client.authorizationToken)
                    }
                }

                val imageBytes: ByteArray = downloadResponse.body()

                val contentType: String = downloadResponse.headers[HttpHeaders.ContentType]!!

                val uri = saveImageToMediaStore(
                    context,
                    imageBytes,
                    imageId.toString(),
                    contentType
                )

                return@async Image(imageId, SyncStatus.BOTH, uri, imageId.toString(), contentType)
            })
        }

        return images
    }

    companion object {
        private const val BASE_URL = "https://api.backblazeb2.com"

        suspend fun createBucket(apiKeyId: String, apiKey: String, name: String): Remote {
            val scope = CoroutineScope(Dispatchers.IO + Job())

            val url: String = "/b2api/v3/b2_create_bucket"

            try {
                val bucketId: String = scope.async(Dispatchers.IO) {
                    val b2Client = getClient(scope, apiKeyId, apiKey).await()

                    try {
                        val response = b2Client.client.post(b2Client.apiUrl + url) {
                            headers {
                                append(HttpHeaders.Authorization, b2Client.authorizationToken)
                            }
                            contentType(ContentType.Application.Json)
                            setBody(
                                buildJsonObject {
                                    put("accountId", b2Client.accountId)
                                    put("bucketName", name)
                                    put("bucketType", "allPrivate")
                                    putJsonObject("defaultServerSideEncryption") {
                                        put("mode", "SSE-B2")
                                        put("algorithm", "AES256")
                                    }
                                }
                            )
                        }

                        val body: JsonElement = response.body()

                        return@async body.jsonObject.getValue("bucketId").jsonPrimitive.content
                    } catch (e: ClientRequestException) {
                        val url = "/b2api/v3/b2_list_buckets"

                        val response = b2Client.client.post(b2Client.apiUrl + url) {
                            headers {
                                append(HttpHeaders.Authorization, b2Client.authorizationToken)
                            }
                            contentType(ContentType.Application.Json)
                            setBody(
                                buildJsonObject {
                                    put("accountId", b2Client.accountId)
                                }
                            )
                        }

                        val body: JsonElement = response.body()

                        val bucketJson =
                            body.jsonObject.getValue("buckets").jsonArray.first { elem ->
                                elem.jsonObject.getValue("bucketName").jsonPrimitive.content == name
                            }

                        return@async bucketJson.jsonObject.getValue("bucketId").jsonPrimitive.content
                    }
                }.await()

                return Remote(name, CloudProviders.BACKBLAZE, apiKeyId, apiKey, bucketId, false)
            } finally {
                scope.cancel()
            }
        }

        private fun getClient(
            scope: CoroutineScope,
            apiKeyId: String,
            apiKey: String
        ): Deferred<B2Client> {
            val url = "/b2api/v3/b2_authorize_account"

            val client = HttpClient(OkHttp) {
                expectSuccess = true
                install(ContentNegotiation) {
                    json(Json {
                        prettyPrint = true
                        isLenient = true
                        ignoreUnknownKeys = true
                    })
                }
                install(HttpTimeout) {
                    requestTimeoutMillis = 60000
                }
                install(HttpRequestRetry) {
                    maxRetries = 3
                    retryOnExceptionIf { request, cause -> cause is ConnectTimeoutException || cause is SocketTimeoutException }
                }
            }

            return scope.async(Dispatchers.IO) {
                try {
                    val response1: HttpResponse = client.get("https://ktor.io/")

                    Log.d("BackBlaze B2", response1.status.value.toString())

                    val response: HttpResponse = client.get(BASE_URL + url) {
                        headers {
                            append(
                                HttpHeaders.Authorization,
                                "Basic" + encodeToBase64("$apiKeyId:$apiKey")
                            )
                        }
                    }

                    val jsonElement: JsonElement = response.body()

                    val authorizationToken =
                        jsonElement.jsonObject.getValue("authorizationToken").jsonPrimitive.content

                    val accountId =
                        jsonElement.jsonObject.getValue("accountId").jsonPrimitive.content

                    val apiUrl =
                        jsonElement
                            .jsonObject.getValue("apiInfo")
                            .jsonObject.getValue("storageApi")
                            .jsonObject.getValue("apiUrl")
                            .jsonPrimitive.content

                    return@async B2Client(authorizationToken, apiUrl, accountId, client)
                } catch (e: Exception) {
                    e.printStackTrace()

                    throw e
                }
            }
        }
    }

    override fun cancel() {
        scope.cancel()
    }
}

class SyncWorker(val appContext: Context, workerParams: WorkerParameters) :
    CoroutineWorker(appContext, workerParams) {

    private val notificationManager =
        applicationContext.getSystemService(NOTIFICATION_SERVICE) as NotificationManager
    private val notificationChannelId = "image_sync_channel"
    private val notificationId = 1

    init {
        createNotificationChannel()
    }

    override suspend fun doWork(): Result = withContext(Dispatchers.IO) {
        try {
            coroutineScope {
                performSynchronization()
            }

            Result.success()
        } catch (e: Exception) {
            Result.failure()
        }
    }

    private suspend fun performSynchronization() {
        val remotesRepository = RemotesRepository(appContext)
        val imagesRepository = MediaRepository(appContext)

        val remote: Remote? = remotesRepository.getPrincipalRemote()

        if (remote == null) return

        val cloudProvider: CloudProvider = remote.getStrategy()

        try {
            val remoteImagesIds = cloudProvider.getRemoteImagesIds()

            val localImages = imagesRepository.getImagesByStatus(SyncStatus.LOCAL)

            val remoteAndLocalImagesIds =
                imagesRepository.getImagesByStatus(SyncStatus.BOTH).map { image -> image.uuid }

            val imagesToDownload = remoteImagesIds.minus(remoteAndLocalImagesIds)

            val totalSyncFiles: Int = imagesToDownload.size + localImages.size

            val uploadedImagesIds = cloudProvider.uploadImages(localImages, appContext)

            val downloadedImages = cloudProvider.downloadImages(imagesToDownload, appContext)

            val uploadJobs: MutableList<Job> = mutableListOf()
            val downloadJobs: MutableList<Job> = mutableListOf()

            var syncedFiles: Int = 0

            fun updateNotification() {
                syncedFiles++

                val messase = "A sincronizar ficheiro $syncedFiles de $totalSyncFiles"
                val progress = (syncedFiles.toFloat() / totalSyncFiles) * 100

                showNotification(progress.toInt(), messase)
            }

            supervisorScope {
                uploadedImagesIds.forEachIndexed { i, imageFuture ->
                    uploadJobs.add(launch(Dispatchers.IO) {
                        val image = imageFuture.await()

                        imagesRepository.updateImageStatus(image, SyncStatus.BOTH)

                        updateNotification()
                    })
                }

                downloadedImages.forEachIndexed { i, imageFuture ->
                    downloadJobs.add(launch(Dispatchers.IO) {
                        val image = imageFuture.await()

                        imagesRepository.addImage(image)

                        updateNotification()
                    })
                }
            }

            uploadJobs.joinAll()
            downloadJobs.joinAll()

            imagesRepository.insertSyncHystory(
                SyncHistory(
                    date = Date(),
                    syncType = SyncType.REMOTE
                )
            )
        } finally {
            cloudProvider.cancel()
        }
    }

    private fun showNotification(progress: Int, message: String) {
        if (ActivityCompat.checkSelfPermission(
                appContext,
                Manifest.permission.POST_NOTIFICATIONS
            ) != PackageManager.PERMISSION_GRANTED
        ) {
            Log.d("SyncWorker", "No permissions for post notification")

            return
        }

        val notification = createNotification(progress, message)

        notificationManager.notify(notificationId, notification.build())
    }

    private fun createNotification(progress: Int, message: String): NotificationCompat.Builder {
        val intent = Intent(appContext, MainActivity::class.java).apply {
            flags = Intent.FLAG_ACTIVITY_NEW_TASK or Intent.FLAG_ACTIVITY_CLEAR_TASK
        }
        val pendingIntent: PendingIntent =
            PendingIntent.getActivity(appContext, 0, intent, PendingIntent.FLAG_IMMUTABLE)

        return NotificationCompat.Builder(appContext, notificationChannelId)
            .setSmallIcon(R.drawable.ic_launcher_foreground) // Replace with your icon
            .setContentTitle("A sincronizar imagens...")
            .setContentText(message)
            .setProgress(100, progress, false)
            .setOngoing(true)
            .setContentIntent(pendingIntent)
    }

    private fun createNotificationChannel() {
        val name = "Images synchronization progress"
        val descriptionText = "Channel for images synchronization progress"
        val importance = NotificationManager.IMPORTANCE_LOW
        val channel = NotificationChannel(notificationChannelId, name, importance).apply {
            description = descriptionText
        }

        notificationManager.createNotificationChannel(channel)
    }
}

核心排查与修复方向 #

1. 后台网络权限配置
   Android 10及以上版本对后台网络请求有默认限制，需在AndroidManifest.xml中为应用添加：

   <application
       ...
       android:usesCleartextTraffic="true">
   

   若需更精准控制，可在Worker中先校验网络可用性：

   val networkManager = appContext.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
   val activeNetwork = networkManager.activeNetwork ?: throw IllegalStateException("无可用网络")
   val capabilities = networkManager.getNetworkCapabilities(activeNetwork)
   if (!capabilities?.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)!!) {
       throw IllegalStateException("网络无互联网访问权限")
   }
   

2. 修复Basic认证格式错误
   getClient中Basic认证头缺少空格，正确格式应为：

   append(
       HttpHeaders.Authorization,
       "Basic " + encodeToBase64("$apiKeyId:$apiKey") // 注意Basic后加空格
   )
   

   缺少空格会导致认证失败，服务器无响应进而触发超时。

3. 明确配置连接超时参数
   当前HttpTimeout仅设置了请求超时，需补充连接超时配置，确保超时参数生效：

   install(HttpTimeout) {
       requestTimeoutMillis = 60000
       connectTimeoutMillis = 15000 // 明确设置连接超时
       socketTimeoutMillis = 60000
   }
   

4. 复用HttpClient与B2Client缓存
   目前每个图片请求都会新建HttpClient并重新调用b2_authorize_account，导致并发连接过载且触发API频率限制。需缓存B2Client实例：

   // 在BackblazeB2类中添加缓存
   private var cachedB2Client: Deferred<B2Client>? = null
   
   private suspend fun getCachedClient(): B2Client {
       return cachedB2Client?.await() ?: run {
           val clientDeferred = getClient(scope, remote.apiKeyId, remote.apiKey)
           cachedB2Client = clientDeferred
           clientDeferred.await()
       }
   }
   

   然后在uploadImages、downloadImages等方法中替换getClient为getCachedClient。

5. 优化OkHttp引擎配置
   为OkHttp配置连接池与后台适配参数，避免连接耗尽：

   val okHttpClient = OkHttpClient.Builder()
       .connectTimeout(15, TimeUnit.SECONDS)
       .readTimeout(60, TimeUnit.SECONDS)
       .writeTimeout(60, TimeUnit.SECONDS)
       .connectionPool(ConnectionPool(5, 1, TimeUnit.MINUTES)) // 调整连接池大小
       .build()
   
   val client = HttpClient(OkHttp) {
       engine {
           preconfigured = okHttpClient
       }
       // 其他配置...
   }
   

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