如何直接将jakarta.ws.rs.core.StreamingOutput传入Jackson的readTree方法？

问题描述 #

我有一个返回jakarta.ws.rs.core.StreamingOutput实体的REST资源方法，这个类可以通过streamingoutput.write(OutputStream os)便捷地向HTTP响应逐步输出内容。我需要在不发起实际HTTP请求的情况下内部复用该方法，现有测试代码可正常运行。问题是：

1. 能否直接将StreamingOutput传入JsonNode jsonRoot = objectMapper.readTree()方法？
2. 有没有无需创建临时String sData/os.toByteArray()字节变量的其他技巧？

附相关代码：

// rest resource function
public Response getList(long custId, String keyword) {
  List<MyEntity> list = JPADB.getMyEntities(custId, keyword);
  StreamingOutput stream = createStream(list);
  CacheControl cc = new CacheControl();
  cc.setNoCache(true);
  return Response.ok().type(TYPE_JSON).cacheControl(cc).entity(stream).build();
}

// somewhere else reuse an existing rest method
MyEntityRest mrest = new MyEntityRest();
Object retvalObj = mrest.getList(1, "test").getEntity();
ByteArrayOutputStream os = new ByteArrayOutputStream(4*1024);           
((StreamingOutput)retvalObj).write(os); 
String sData = os.toString("UTF-8");
JsonNode jsonRoot = om.readTree(sData);
JsonNode jsonItem = jsonRoot.path("items").get(0);
System.out.println(String.format("%s %s"
   , jsonItem.path("code").asText("")
   , jsonItem.path("names").get(0).path("value").asText("")
));

解决方案 #

关于第一个问题 #

不能直接将StreamingOutput传入objectMapper.readTree()，因为Jackson的readTree()方法没有提供接收StreamingOutput类型的重载，它只支持File、InputStream、Reader、String等类型的输入。

无需临时变量的替代方案 #

方案1：使用管道流（适合大数据量，无内存缓存）

用PipedInputStream和PipedOutputStream搭建管道，让StreamingOutput直接向管道写数据，同时Jackson从管道读数据解析，避免把整个内容加载到内存：

MyEntityRest mrest = new MyEntityRest();
StreamingOutput stream = (StreamingOutput) mrest.getList(1, "test").getEntity();

// 创建管道流对
PipedOutputStream pos = new PipedOutputStream();
PipedInputStream pis = new PipedInputStream(pos);

// 启动线程执行StreamingOutput的写入操作（管道流读写阻塞，必须异步）
new Thread(() -> {
    try {
        stream.write(pos);
        pos.close();
    } catch (IOException e) {
        throw new RuntimeException("写入管道流失败", e);
    }
}).start();

// 直接从管道输入流解析JsonNode
JsonNode jsonRoot = om.readTree(pis);
pis.close();

// 后续处理逻辑不变
JsonNode jsonItem = jsonRoot.path("items").get(0);
System.out.println(String.format("%s %s"
       , jsonItem.path("code").asText("")
       , jsonItem.path("names").get(0).path("value").asText("")
));

方案2：抽离序列化逻辑（最优，避免重复序列化反序列化）

既然StreamingOutput是从List<MyEntity>序列化而来，不如直接复用序列化逻辑，跳过StreamingOutput这一层，直接从原数据生成JsonNode，这是性能最好、代码最简洁的方案：

首先抽离通用序列化方法：

// 抽离List转JsonNode的逻辑
public JsonNode listToJsonNode(List<MyEntity> list) throws JsonProcessingException {
    // 如果createStream里是用Jackson标准序列化，直接用valueToTree即可
    return om.valueToTree(list);
    
    // 如果createStream里是自定义的JsonGenerator序列化逻辑，可以改用TreeGenerator：
    // ObjectNode rootNode = om.createObjectNode();
    // JsonGenerator gen = om.getFactory().createGenerator(new TreeGenerator(rootNode));
    // // 执行createStream里的自定义序列化逻辑
    // gen.close();
    // return rootNode;
}

然后在REST方法和内部复用场景分别使用：

// REST资源方法保持原逻辑
public Response getList(long custId, String keyword) {
  List<MyEntity> list = JPADB.getMyEntities(custId, keyword);
  StreamingOutput stream = createStream(list);
  CacheControl cc = new CacheControl();
  cc.setNoCache(true);
  return Response.ok().type(TYPE_JSON).cacheControl(cc).entity(stream).build();
}

// 内部复用场景直接调用抽离的方法
List<MyEntity> list = JPADB.getMyEntities(1, "test");
JsonNode jsonRoot = listToJsonNode(list);
// 后续处理逻辑不变

方案3：自定义InputStream包装（单线程，仍需内存缓存）

如果不想开线程，也可以自定义InputStream包装StreamingOutput，省去手动转换String的步骤，但本质还是把数据加载到内存：

public class StreamingOutputInputStream extends InputStream {
    private final InputStream innerStream;

    public StreamingOutputInputStream(StreamingOutput output) throws IOException {
        ByteArrayOutputStream buffer = new ByteArrayOutputStream();
        output.write(buffer);
        innerStream = new ByteArrayInputStream(buffer.toByteArray());
    }

    @Override
    public int read() throws IOException {
        return innerStream.read();
    }
}

使用方式：

StreamingOutput stream = (StreamingOutput) mrest.getList(1, "test").getEntity();
JsonNode jsonRoot = om.readTree(new StreamingOutputInputStream(stream));
// 后续处理逻辑不变

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