如何用Retrofit2解析每行一个JSON对象的流式API响应？

你的问题根源在于：API返回的是每行一个独立JSON对象的流格式，而非标准的JSON数组，Gson默认转换器仅能处理单个JSON对象或完整JSON数组，因此会出现解析错误。以下两种方案可实现自动转换为对象列表：

方案一：自定义Retrofit ConverterFactory（推荐） #

通过自定义转换器实现逐行读取流并解析JSON对象，最终返回对象列表，适配Retrofit的调用逻辑。

1. 实现自定义Converter和ConverterFactory #

public class LineDelimitedGsonConverterFactory extends Converter.Factory {
    private final Gson gson;

    public static LineDelimitedGsonConverterFactory create(Gson gson) {
        return new LineDelimitedGsonConverterFactory(gson);
    }

    private LineDelimitedGsonConverterFactory(Gson gson) {
        this.gson = gson;
    }

    @Override
    public Converter<ResponseBody, ?> responseBodyConverter(Type type, Annotation[] annotations, Retrofit retrofit) {
        // 仅处理List类型的返回值
        if (!(type instanceof ParameterizedType) || ((ParameterizedType) type).getRawType() != List.class) {
            return super.responseBodyConverter(type, annotations, retrofit);
        }
        Type itemType = ((ParameterizedType) type).getActualTypeArguments()[0];
        return new LineDelimitedResponseBodyConverter<>(gson, itemType);
    }

    private static class LineDelimitedResponseBodyConverter<T> implements Converter<ResponseBody, List<T>> {
        private final Gson gson;
        private final Type itemType;

        public LineDelimitedResponseBodyConverter(Gson gson, Type itemType) {
            this.gson = gson;
            this.itemType = itemType;
        }

        @Override
        public List<T> convert(ResponseBody value) throws IOException {
            List<T> result = new ArrayList<>();
            BufferedReader reader = new BufferedReader(new InputStreamReader(value.byteStream()));
            String line;
            try {
                while ((line = reader.readLine()) != null) {
                    // 跳过空行
                    if (line.trim().isEmpty()) continue;
                    T item = gson.fromJson(line, itemType);
                    result.add(item);
                }
            } finally {
                value.close();
            }
            return result;
        }
    }
}

2. 修改Retrofit配置 #

将自定义转换器放在GsonConverterFactory之前，确保优先处理流式响应：

Gson gson = new GsonBuilder()
        .setFieldNamingPolicy(FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES)
        .create();

Retrofit retrofit = new Retrofit.Builder()
        .baseUrl(BASE_URL)
        .client(okHttpClient)
        .addConverterFactory(LineDelimitedGsonConverterFactory.create(gson))
        .addConverterFactory(GsonConverterFactory.create(gson))
        .build();

3. 修改API接口定义 #

直接声明返回Call<List<Response>>：

public interface Api {
    @Streaming
    @POST("/stream")
    @Headers({"accept: application/stream+json", "content-type: application/json"})
    Call<List<Response>> generateStream2(@Body Request request);
}

方案二：手动处理ResponseBody流 #

若不想自定义转换器，可直接在接口中返回ResponseBody，在回调内手动逐行解析：

1. 修改API接口 #

public interface Api {
    @Streaming
    @POST("/stream")
    @Headers({"accept: application/stream+json", "content-type: application/json"})
    Call<ResponseBody> generateStream2(@Body Request request);
}

2. 调用时处理流 #

api.generateStream2(request).enqueue(new Callback<ResponseBody>() {
    @Override
    public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
        if (!response.isSuccessful()) {
            // 处理请求错误
            return;
        }
        BufferedReader reader = new BufferedReader(new InputStreamReader(response.body().byteStream()));
        List<Response> responseList = new ArrayList<>();
        String line;
        try {
            while ((line = reader.readLine()) != null) {
                if (line.trim().isEmpty()) continue;
                Response item = gson.fromJson(line, Response.class);
                responseList.add(item);
            }
            // 处理解析完成的对象列表
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                response.body().close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    @Override
    public void onFailure(Call<ResponseBody> call, Throwable t) {
        // 处理请求失败
    }
});

关键注意事项 #

• 必须保留@Streaming注解，避免OkHttp将整个流加载到内存，处理大流量响应时更高效。
• 自定义转换器方案更贴合Retrofit的设计逻辑，无需在业务代码中重复编写流解析逻辑。

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