如何使用Retrofit2从API读取连续流式数据？

要处理这种持续传输的海量非JSON流数据，核心思路是绕开Retrofit的默认转换器，直接获取原始响应体的输入流来逐块读取数据，具体步骤如下：

1. 定义返回ResponseBody的API接口 #

在Retrofit接口中，让@GET方法返回ResponseBody而非自定义实体类，这样可以拿到原始的响应流：

public interface StreamingApi {
    @GET("/stream")
    Call<ResponseBody> getStreamingData();
}

如果用Kotlin协程，定义成挂起函数：

interface StreamingApi {
    @GET("/stream")
    suspend fun getStreamingData(): ResponseBody
}

2. 执行请求并持续读取流数据 #

调用接口后，通过ResponseBody获取输入流，循环读取数据直到流结束。注意要在后台线程处理，避免阻塞主线程：

Java示例（Callback方式） #

Retrofit retrofit = new Retrofit.Builder()
        .baseUrl("https://example")
        .build();

StreamingApi api = retrofit.create(StreamingApi.class);
Call<ResponseBody> call = api.getStreamingData();

call.enqueue(new Callback<ResponseBody>() {
    @Override
    public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
        if (!response.isSuccessful() || response.body() == null) {
            // 处理请求失败或空响应
            return;
        }

        try (InputStream inputStream = response.body().byteStream();
             BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream))) {
            
            String line;
            // 持续读取每一行数据（根据流格式调整读取逻辑）
            while ((line = reader.readLine()) != null) {
                processStreamingData(line);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void onFailure(Call<ResponseBody> call, Throwable t) {
        t.printStackTrace();
    }
});

// 自定义数据处理方法
private void processStreamingData(String rawData) {
    // 根据流的实际格式解析数据，比如分割字段、提取有效内容等
    System.out.println("Received chunk: " + rawData);
}

Kotlin协程示例 #

val retrofit = Retrofit.Builder()
    .baseUrl("https://example")
    .build()

val api = retrofit.create(StreamingApi::class.java)

CoroutineScope(Dispatchers.IO).launch {
    try {
        val responseBody = api.getStreamingData()
        responseBody.byteStream().use { inputStream ->
            BufferedReader(InputStreamReader(inputStream)).useLines { lines ->
                lines.forEach { line ->
                    processStreamingData(line)
                }
            }
        }
    } catch (e: Exception) {
        e.printStackTrace()
    }
}

// 自定义数据处理函数
private fun processStreamingData(rawData: String) {
    println("Received chunk: $rawData")
}

3. 关键注意事项 #

• 禁用JSON转换器：不要添加GsonConverterFactory等默认转换器，否则Retrofit会尝试一次性解析整个响应，导致解析失败或内存溢出。
• 资源管理：务必用try-with-resources（Java）或use函数（Kotlin）自动关闭输入流，避免资源泄漏。
• 线程处理：流读取操作耗时较长，必须放在后台线程执行，Android环境下可以用Coroutine的Dispatchers.IO或RxJava的io()调度器。

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