如何使用Jackson JsonParser多线程逐对象读取输入流JSON数据

JsonParser是有状态的非线程安全组件，内部维护了字节缓冲区、流读取偏移量、当前token解析状态等可变数据，同时对接的单个顺序传输输入流本身就不支持多线程并发随机读取。多个线程直接并发调用同一个parser实例的nextToken()方法，会直接打乱内部状态指针，出现跳过条目、解析结构错乱、抛出异常都是必然结果，不存在任何让多线程直接操作同一个输入流/parser实例还能保证数据正确性的方案。

采用单线程解析拆分 + 多线程消费处理的生产者-消费者架构，从架构层面避免多线程接触共享的有状态解析组件，全程不会丢失任何JSON条目：

• 单独启动1个解析线程，唯一持有输入流和JsonParser实例，这个线程只做顺序解析工作：逐次调用nextToken()从流中读取数据，每识别到一个完整的JSON对象，就将其解析为和parser状态完全解绑的独立对象，整个过程不做重业务逻辑，避免阻塞流读取
• 初始化有界阻塞队列作为解析线程和工作线程的中转缓冲区，队列容量根据业务处理速度和内存上限设置，避免无界队列导致OOM
• 启动固定大小的工作线程池，所有工作线程仅从阻塞队列中获取已经解析完成的独立JSON对象，执行后续业务逻辑，全程不接触原始输入流和parser实例
• 输入流读取到结束位置后，解析线程向队列中放入结束标识，等工作线程将队列中剩余的所有JSON条目处理完成后，再关闭线程池，保证所有条目都被消费

• 不要尝试给nextToken()方法加锁实现多线程调用：锁本质是把并发操作退化为串行执行，不仅没有性能提升，还会因为线程上下文切换、拿锁顺序不确定带来额外开销和逻辑乱序问题，完全没有必要
• 解析出的JSON对象必须是完全独立的副本：不要直接传递parser引用或者绑定parser当前状态的临时对象，否则parser移动读取位置后，临时对象的内容会被覆盖，导致数据错乱。如果用Jackson解析，直接调用readValueAsTree()转成JsonNode，或者绑定到对应POJO类即可，这两类对象和parser状态完全解耦，可以安全跨线程传递
• 队列要做流量控制：当工作线程处理速度跟不上解析速度时，有界队列满了之后会自动阻塞解析线程，实现自然反压，避免内存溢出
• 结束标识需要特殊处理：单个结束标识只能被一个工作线程拿到，拿到标识的线程需要把标识重新放回队列，保证所有工作线程都能感知到解析结束，避免线程永久阻塞

JsonFactory jsonFactory = new JsonFactory();
// 初始化解析器、有界队列、结束标识
try (JsonParser parser = jsonFactory.createParser(yourInputStream)) {
    int queueCapacity = 1000;
    BlockingQueue<Object> processQueue = new ArrayBlockingQueue<>(queueCapacity);
    Object finishFlag = new Object();
    int workerCount = Runtime.getRuntime().availableProcessors();
    ExecutorService workerPool = Executors.newFixedThreadPool(workerCount);

    // 启动工作线程
    for (int i = 0; i < workerCount; i++) {
        workerPool.submit(() -> {
            try {
                while (true) {
                    Object item = processQueue.take();
                    if (item == finishFlag) {
                        // 把结束标识放回队列，让其他工作线程也能收到结束信号
                        processQueue.put(finishFlag);
                        break;
                    }
                    // 替换为实际的业务处理逻辑
                    handleSingleJson((JsonNode) item);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
    }

    // 单线程顺序解析流
    // 如果你的流是连续输出独立JSON对象而非JSON数组，可跳过下一行的数组起始位读取
    parser.nextToken();
    while (parser.nextToken() == JsonToken.START_OBJECT) {
        JsonNode currentNode = parser.readValueAsTree();
        processQueue.put(currentNode);
    }

    // 所有条目解析完成，放入结束标识
    processQueue.put(finishFlag);
    // 等待所有任务处理完成后关闭线程池
    workerPool.shutdown();
    workerPool.awaitTermination(1, TimeUnit.HOURS);
}

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