如何用Gson将JSON数组懒反序列化为Stream<JsonObject>

问题背景 #

假设有一个大型（甚至可能无限）的JSON对象数组：

[
    {
        "id": 41,
        "name": "foo"
    },
    {
        "id": 42,
        "name": "bar"
    },
    {
        "id": 43,
        "name": "baz"
    }
]

在JVM应用中，可通过java.io.Reader实例访问该JSON数据。常规方式是用Gson将其反序列化为JsonObject列表，但这种方式会把所有数据加载到内存中，对于大型或无限数据来说内存开销极大：

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.reflect.TypeToken;

import java.io.Reader;
import java.util.List;

class C {
    List<JsonObject> readAsList(final Reader in) {
        final var gson = new Gson();
        final var token = new TypeToken<List<JsonObject>>() {};

        return gson.fromJson(in, token);
    }
}

我们需要实现懒加载的冷流（Stream），避免一次性加载所有数据到内存容器中。

解决方案1：手动遍历JsonReader构建懒加载流 #

Gson的JsonReader提供了低层级的JSON解析能力，可以逐个读取数组元素，基于它构建Stream<JsonObject>就能实现懒加载：

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.JsonReader;
import com.google.gson.stream.JsonToken;

import java.io.Reader;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

class JsonStreamReader {
    private final Gson gson = new Gson();

    Stream<JsonObject> readAsStream(final Reader reader) {
        JsonReader jsonReader = gson.newJsonReader(reader);
        try {
            // 确认当前是JSON数组的起始位置
            jsonReader.beginArray();
        } catch (Exception e) {
            throw new RuntimeException("读取JSON数组起始位置失败", e);
        }

        // 创建Spliterator，逐个读取并反序列化数组元素
        Spliterator<JsonObject> spliterator = Spliterators.spliteratorUnknownSize(
                new java.util.Iterator<>() {
                    @Override
                    public boolean hasNext() {
                        try {
                            JsonToken nextToken = jsonReader.peek();
                            if (nextToken == JsonToken.END_ARRAY) {
                                jsonReader.close();
                                return false;
                            }
                            return nextToken != JsonToken.END_DOCUMENT;
                        } catch (Exception e) {
                            throw new RuntimeException("检查下一个元素失败", e);
                        }
                    }

                    @Override
                    public JsonObject next() {
                        try {
                            // 反序列化单个JsonObject
                            return gson.fromJson(jsonReader, JsonObject.class);
                        } catch (Exception e) {
                            throw new RuntimeException("反序列化JsonObject失败", e);
                        }
                    }
                },
                Spliterator.ORDERED | Spliterator.NONNULL
        );

        // 将Spliterator包装为Stream，通过onClose确保资源释放
        return StreamSupport.stream(spliterator, false)
                .onClose(() -> {
                    try {
                        jsonReader.close();
                    } catch (Exception e) {
                        throw new RuntimeException("关闭JsonReader失败", e);
                    }
                });
    }
}

核心特性 #

• 完全懒加载：仅当Stream执行终端操作（如forEach、collect）时，才会逐个读取并反序列化JSON元素
• 通过Spliterator将迭代器转换为Stream，保留流的特性
• 强制在Stream关闭时释放JsonReader资源，避免泄漏

解决方案2：自定义TypeAdapter适配Stream #

利用Gson的扩展机制，自定义TypeAdapterFactory，让Gson直接支持将JSON数组反序列化为Stream<JsonObject>：

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.TypeAdapter;
import com.google.gson.TypeAdapterFactory;
import com.google.gson.reflect.TypeToken;
import com.google.gson.stream.JsonReader;
import com.google.gson.stream.JsonToken;

import java.io.Reader;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

class StreamTypeAdapterFactory implements TypeAdapterFactory {
    @SuppressWarnings("unchecked")
    @Override
    public <T> TypeAdapter<T> create(Gson gson, TypeToken<T> type) {
        // 仅处理Stream<JsonObject>类型
        if (type.getRawType() == Stream.class && type.getType() instanceof java.lang.reflect.ParameterizedType paramType) {
            Class<?> elementType = (Class<?>) paramType.getActualTypeArguments()[0];
            if (elementType == JsonObject.class) {
                TypeAdapter<JsonObject> elementAdapter = gson.getAdapter(JsonObject.class);
                return (TypeAdapter<T>) new StreamTypeAdapter<>(elementAdapter);
            }
        }
        return null;
    }

    private static class StreamTypeAdapter<E> extends TypeAdapter<Stream<E>> {
        private final TypeAdapter<E> elementAdapter;

        StreamTypeAdapter(TypeAdapter<E> elementAdapter) {
            this.elementAdapter = elementAdapter;
        }

        @Override
        public void write(com.google.gson.stream.JsonWriter out, Stream<E> value) {
            throw new UnsupportedOperationException("暂不支持写入Stream类型");
        }

        @Override
        public Stream<E> read(JsonReader in) {
            try {
                in.beginArray();
            } catch (Exception e) {
                throw new RuntimeException("读取数组起始位置失败", e);
            }

            Spliterator<E> spliterator = Spliterators.spliteratorUnknownSize(
                    new java.util.Iterator<>() {
                        @Override
                        public boolean hasNext() {
                            try {
                                JsonToken token = in.peek();
                                if (token == JsonToken.END_ARRAY) {
                                    in.close();
                                    return false;
                                }
                                return token != JsonToken.END_DOCUMENT;
                            } catch (Exception e) {
                                throw new RuntimeException("检查下一个元素失败", e);
                            }
                        }

                        @Override
                        public E next() {
                            try {
                                return elementAdapter.read(in);
                            } catch (Exception e) {
                                throw new RuntimeException("反序列化元素失败", e);
                            }
                        }
                    },
                    Spliterator.ORDERED | Spliterator.NONNULL
            );

            return StreamSupport.stream(spliterator, false)
                    .onClose(() -> {
                        try {
                            in.close();
                        } catch (Exception e) {
                            throw new RuntimeException("关闭JsonReader失败", e);
                        }
                    });
        }
    }
}

// 使用示例
class StreamReaderExample {
    public static void main(String[] args) {
        Gson gson = new Gson.Builder()
                .registerTypeAdapterFactory(new StreamTypeAdapterFactory())
                .create();

        try (Reader reader = ...; // 替换为你的Reader实例
             Stream<JsonObject> stream = gson.fromJson(reader, new TypeToken<Stream<JsonObject>>() {}.getType())) {

            stream.forEach(obj -> {
                // 处理单个JsonObject
                System.out.println("ID: " + obj.get("id").getAsInt() + ", Name: " + obj.get("name").getAsString());
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

核心特性 #

• 符合Gson扩展规范，代码复用性更强
• 同样实现懒加载，仅在流操作触发时解析数据
• 可以直接通过gson.fromJson获取Stream，使用方式更贴近常规Gson操作

注意事项 #

• 必须使用try-with-resources语法确保Stream被正确关闭，避免JsonReader资源泄漏
• 数据源Reader必须保持可用直到流处理完成，不能提前关闭
• 生成的Stream是冷流，仅当执行终端操作时才会开始解析JSON

内容的提问来源于stack exchange，提问作者Андрей Щеглов