Akka Flow中用javax.xml.transform.Transformer转XML遇超时问题求助

问题场景 #

尝试在Akka Flow中用javax.xml.transform.Transformer做XML转换，编写了如下代码：

Flow<ByteString, ByteString, NotUsed> transformXml(@NonNull final Source<ByteString, ?> xslt) {
    final Pair<InputStream, Sink<ByteString, NotUsed>> input = Flow
        .of(ByteString.class)
        .toMat(StreamConverters.asInputStream(), Keep.right())
        .preMaterialize(actorSystem);
    final javax.xml.transform.Source xmlSource = new StreamSource(input.first());
    final javax.xml.transform.Source xsltSource = new StreamSource(xslt.runWith(StreamConverters.asInputStream(), actorSystem));
    final Source<ByteString, NotUsed> transformed = Source.lazySource(() -> {
      final Pair<OutputStream, Source<ByteString, NotUsed>> pair = StreamConverters
          .asOutputStream(Duration.ofMinutes(5))
          .preMaterialize(actorSystem);
      final OutputStream outputStream = pair.first();
      final Result output = new StreamResult(outputStream);
      final TransformerFactory transformerFactory = TransformerFactory.newInstance();
      final Transformer transformer = transformerFactory.newTransformer(xsltSource);
      final Source<ByteString, NotUsed> transformerStream = pair.second();
      transformer.transform(xmlSource, output);
      outputStream.close();
      return transformerStream;
    }).mapMaterializedValue(i -> NotUsed.getInstance()).mapError(TransformerException.class, transformerException -> transformerException);
    return Flow.fromSinkAndSource(input.second(), transformed);
}

运行测试代码时立即触发超时，transformer#transform出现阻塞：

@Test
void transformFlow() {
    final String expectedResult = "...";
    final Source<ByteString, NotUsed> xmlSource = Source.single(ByteString.fromString("some valid xml"));
    final Source<ByteString, NotUsed> xsltSource = Source.single(ByteString.fromString("some valid xslt"));
    Assertions.assertEquals(expectedResult,
        xmlSource
            .via(transformXml(xsltSource))
            .runWith(akka.stream.javadsl.Sink.fold(ByteString.emptyByteString(), ByteString::concat), actorSystem)
            .toCompletableFuture()
            .join()
            .utf8String());
}

报错信息：

javax.xml.transform.TransformerException: javax.xml.transform.TransformerException: com.sun.org.apache.xml.internal.utils.WrappedRuntimeException: Timeout on waiting for new data

原因分析 #

1. 阻塞调用与流预物化的冲突：transformer.transform是同步阻塞方法，但代码中提前预物化了XML输入的InputStream（通过preMaterialize），此时还没有任何XML数据流入该输入流，导致transformer尝试读取时直接触发超时。
2. 违反Akka Stream懒加载语义：直接调用xslt.runWith(StreamConverters.asInputStream(), actorSystem)会提前启动XSLT流的物化，而非在转换逻辑执行时按需加载，破坏了流的生命周期管理。
3. 阻塞线程占用Akka调度线程：在Source.lazySource内部执行同步阻塞的转换逻辑，会占用Akka的调度线程，不仅违反了Akka Stream的非阻塞设计原则，还会导致流的其他阶段无法正常调度。

修复方案 #

针对上述问题，调整代码逻辑，确保流的生命周期正确，阻塞操作隔离到专用线程池：

修复后的代码 #

import akka.NotUsed;
import akka.stream.javadsl.Flow;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;
import akka.util.ByteString;
import javax.xml.transform.Transformer;
import javax.xml.transform.TransformerFactory;
import javax.xml.transform.stream.StreamResult;
import javax.xml.transform.stream.StreamSource;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class XmlTransformerFlow {

    // 专用线程池处理阻塞的XML转换逻辑，避免占用Akka调度线程
    private static final Executor TRANSFORMER_EXECUTOR = Executors.newFixedThreadPool(4);

    public Flow<ByteString, ByteString, NotUsed> transformXml(Source<ByteString, ?> xsltSource) {
        // 提前加载XSLT内容（XSLT通常体积较小，适合一次性加载复用）
        final ByteString xsltBytes = xsltSource
                .runWith(Sink.fold(ByteString.empty(), ByteString::concat), actorSystem)
                .toCompletableFuture()
                .join();

        return Flow.of(ByteString.class)
                // 收集全部XML输入（JAXP Transformer默认基于DOM，需要完整XML内容）
                .fold(ByteString.empty(), ByteString::concat)
                // 异步执行阻塞转换，指定专用线程池
                .mapAsync(1, xmlBytes -> {
                    return actorSystem.dispatcher().submit(() -> {
                        try (ByteArrayInputStream xmlIn = new ByteArrayInputStream(xmlBytes.toArray());
                             ByteArrayOutputStream xmlOut = new ByteArrayOutputStream()) {

                            StreamSource xmlStreamSource = new StreamSource(xmlIn);
                            StreamSource xsltStreamSource = new StreamSource(new ByteArrayInputStream(xsltBytes.toArray()));
                            StreamResult result = new StreamResult(xmlOut);

                            TransformerFactory factory = TransformerFactory.newInstance();
                            Transformer transformer = factory.newTransformer(xsltStreamSource);
                            transformer.transform(xmlStreamSource, result);

                            return ByteString.fromArray(xmlOut.toByteArray());
                        } catch (Exception e) {
                            throw new RuntimeException("XML转换失败", e);
                        }
                    }, TRANSFORMER_EXECUTOR);
                })
                // 将转换后的单ByteString展开为Flow输出
                .flatMapConcat(Source::single);
    }
}

关键修复点 #

1. 预加载XSLT内容：一次性读取XSLT的全部内容并缓存，避免重复加载，符合XSLT静态复用的场景（如果需要动态XSLT，可调整为每次转换时加载）。
2. 收集完整XML输入：JAXP的Transformer默认基于DOM解析，需要完整的XML内容才能正确转换；若需处理超大XML，可改用StAX流式API配合分块处理。
3. 异步隔离阻塞操作：使用mapAsync配合专用线程池执行转换逻辑，避免阻塞Akka的调度线程，符合Akka Stream的非阻塞设计原则。
4. 移除提前预物化操作：所有流操作都在Flow的处理阶段执行，确保数据流入后才触发转换，解决了输入流空等待的超时问题。

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