在Android Studio中如何连接URL持续数据流并处理非标准JSON对象

你用的AsyncHttpClient库已停止维护多年，对长连接、流式响应的支持很差，默认的TextHttpResponseHandler会等待全部响应内容返回后才触发回调，既会因为长连接读超时导致连接失败，也无法实现逐段处理数据流的需求。你遇到的流格式属于JSON Lines（又称NDJSON，换行分隔JSON），每个独立JSON对象为一条数据，整体不需要是合法的JSON数组，适配逻辑很简单，建议直接用你已经熟悉的OkHttp/Retrofit生态实现。

方案1：OkHttp 流式读取 + 增量解析 #

无需引入额外依赖，Retrofit底层默认依赖OkHttp，直接使用即可：

步骤1：配置OkHttpClient，适配长连接场景 #

// Kotlin 示例，Java语法逻辑完全一致
val okHttpClient = OkHttpClient.Builder()
    .connectTimeout(30, TimeUnit.SECONDS)
    .readTimeout(0, TimeUnit.SECONDS) // 关闭读超时，避免长连接主动断开
    .build()

步骤2：构建请求、发起异步调用 #

// 构造带参数的请求URL
val requestUrl = HttpUrl.get("你的接口完整地址")
    .newBuilder()
    .addQueryParameter("limit", "5")
    .addQueryParameter("page", "0")
    .build()

val request = Request.Builder()
    .url(requestUrl)
    .build()

// 发起异步请求
okHttpClient.newCall(request).enqueue(object : Callback {
    override fun onFailure(call: Call, e: IOException) {
        Log.e(TAG, "请求失败：${e.message}")
    }

    override fun onResponse(call: Call, response: Response) {
        if (!response.isSuccessful) {
            Log.e(TAG, "接口返回错误，状态码：${response.code}")
            return
        }
        val inputStream = response.body()?.byteStream() ?: return
        parseStream(inputStream)
    }
})

步骤3：增量解析流数据 #

这里提供两种解析方式，根据你拿到的流格式选择：

• 如果JSON对象严格按行分隔，用逐行读取的方式：

private fun parseStream(inputStream: InputStream) {
    val reader = BufferedReader(InputStreamReader(inputStream, Charsets.UTF_8))
    val gson = Gson()
    // 先定义对应单条JSON结构的实体类
    data class User(
        val id: String,
        val name: String
    )
    data class StreamItem(
        val to: User,
        val from: User,
        val timestamp: String,
        val alive: Boolean
    )

    var line: String?
    while (reader.readLine().also { line = it } != null) {
        line?.takeIf { it.isNotBlank() }?.let { jsonStr ->
            try {
                val item = gson.fromJson(jsonStr, StreamItem::class.java)
                // 此处已拿到单条解析后的数据，切换到主线程即可更新UI
                Log.d(TAG, "解析到数据：${item.from.name} -> ${item.to.name}")
            } catch (e: JsonSyntaxException) {
                // 跳过格式错误的行即可
                Log.w(TAG, "无效JSON行：$jsonStr", e)
            }
        }
    }
    reader.close()
}

• 如果JSON对象之间没有严格换行、存在空白字符，用Gson流式解析器适配，兼容性更强：

private fun parseStream(inputStream: InputStream) {
    val jsonReader = JsonReader(InputStreamReader(inputStream, Charsets.UTF_8))
    jsonReader.isLenient = true // 允许非标准JSON格式的流输入
    val gson = Gson()
    // 实体类定义和上面一致
    data class User(
        val id: String,
        val name: String
    )
    data class StreamItem(
        val to: User,
        val from: User,
        val timestamp: String,
        val alive: Boolean
    )

    while (jsonReader.peek() != JsonToken.END_DOCUMENT) {
        try {
            val item = gson.fromJson<StreamItem>(jsonReader, StreamItem::class.java)
            // 处理解析后的单条数据
            Log.d(TAG, "解析到数据：${item.from.name} -> ${item.to.name}")
        } catch (e: Exception) {
            // 跳过解析错误的条目
            jsonReader.skipValue()
            Log.w(TAG, "条目解析失败", e)
        }
    }
    jsonReader.close()
}

方案2：Retrofit 适配方案 #

如果你想保持Retrofit的接口定义风格，只需在接口方法上加@Streaming注解即可：

1. 定义接口

public interface StreamApiService {
    @Streaming
    @GET("你的接口路径")
    Call<ResponseBody> getStreamData(@Query("limit") int limit, @Query("page") int page);
}

2. 调用接口拿到ResponseBody后，用和上面完全相同的parseStream方法解析即可。

• 所有网络请求、流读取、JSON解析操作都在子线程执行，拿到数据后必须通过Handler、LiveData或者协程切换到主线程再更新UI，否则会触发Android的线程检查报错。
• 如果流是无限长连接，不需要时主动调用call.cancel()即可断开连接，避免资源泄漏。

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