java 中多个 json 到 xml 的高效处理

Question

我需要处理存储在文件中的多个 json 并将它们转换为 xml。这些 json 将平均转换为 3000 行 xml（以获得 json 大小的图片）。我希望在不到半小时内实现这一目标。

时间包括从文件中读取、遍历和转换并再次存储到另一个文件中作为xml。稍后我还需要将它们存储在数据库中。我还没有这样做，但我打算使用 Kafka 连接器直接插入数据库以实现性能。

我面临的问题是：- - 我只能从我的输入文本文件中读取一行，但不能读取多行。 - 如何在接受要转换的字符串的 SpliteratorBenchmark.processLine() 方法中使用我的实用程序。 - 如何在每次创建新的 xml 文件时创建一个新的 xml 文件。我有一个包含多个 json 的文件。我应该读取一行并将其转换为 xml 然后另一个 xml 用于下一个 json.

请帮忙。

有没有其他方法可以做到这一点。我不能为此任务使用任何类型的 JMS，例如 MQ 或 Kafka。

这是我的核心 FixedBatchSpliteratorBase:-

public abstract class FixedBatchSpliteratorBase<T> implements Spliterator<T> {
      private final int batchSize;
      private final int characteristics;
      private long est;

      public FixedBatchSpliteratorBase(int characteristics, int batchSize, long est) {
        this.characteristics = characteristics | SUBSIZED;
        this.batchSize = batchSize;
        this.est = est;
      }
      public FixedBatchSpliteratorBase(int characteristics, int batchSize) {
        this(characteristics, batchSize, Long.MAX_VALUE);
      }
      public FixedBatchSpliteratorBase(int characteristics) {
        this(characteristics, 128, Long.MAX_VALUE);
      }
      public FixedBatchSpliteratorBase() {
        this(IMMUTABLE | ORDERED | NONNULL);
      }

      @Override 
      public Spliterator<T> trySplit() {
        final HoldingConsumer<T> holder = new HoldingConsumer<T>();
        if (!tryAdvance(holder)) return null;
        final Object[] a = new Object[batchSize];
        int j = 0;
        do a[j] = holder.value; while (++j < batchSize && tryAdvance(holder));
        if (est != Long.MAX_VALUE) est -= j;
        return spliterator(a, 0, j, characteristics() | SIZED);
      }


      @Override 
      public Comparator<? super T> getComparator() {
        if (hasCharacteristics(SORTED)) return null;
        throw new IllegalStateException();
      }

      @Override 
      public long estimateSize() { 
          return est; 
      }

      @Override 
      public int characteristics() { 
          return characteristics; 
      }

      static final class HoldingConsumer<T> implements Consumer<T> {
        Object value;

        @Override 
        public void accept(T value) { 
            this.value = value; 
        }
      }
    }

这是我的 class FixedBatchSpliterator:-

import static java.util.stream.StreamSupport.stream;

import java.util.Spliterator;
import java.util.function.Consumer;
import java.util.stream.Stream;

public class FixedBatchSpliterator<T> extends FixedBatchSpliteratorBase<T> {
  private final Spliterator<T> spliterator;

  public FixedBatchSpliterator(Spliterator<T> toWrap, int batchSize) {
    super(toWrap.characteristics(), batchSize, toWrap.estimateSize());
    this.spliterator = toWrap;
  }

  public static <T> FixedBatchSpliterator<T> batchedSpliterator(Spliterator<T> toWrap, int batchSize) {
    return new FixedBatchSpliterator<>(toWrap, batchSize);
  }

  public static <T> Stream<T> withBatchSize(Stream<T> in, int batchSize) {
    return stream(batchedSpliterator(in.spliterator(), batchSize), true);
  }

  @Override public boolean tryAdvance(Consumer<? super T> action) {
    return spliterator.tryAdvance(action);
  }
  @Override public void forEachRemaining(Consumer<? super T> action) {
    spliterator.forEachRemaining(action);
  }
}

这是我的 JsonSpliterator:-

import java.util.function.Consumer;

import com.dj.gls.core.FixedBatchSpliteratorBase;


public class JsonSpliterator extends FixedBatchSpliteratorBase<String[]> {
    private final JSONReader jr;

    JsonSpliterator(JSONReader jr, int batchSize) {
      super(IMMUTABLE | ORDERED | NONNULL, batchSize);
      if (jr == null) throw new NullPointerException("JSONReader is null");
      this.jr = jr;
    }
    public JsonSpliterator(JSONReader jr) { this(jr, 128); }

    @Override 
    public boolean tryAdvance(Consumer<? super String[]> action) {
      if (action == null) throw new NullPointerException();
      try {
        final String[] row = jr.readNext();
        if (row == null) return false;
        action.accept(row);
        return true;
      } catch (RuntimeException e) {  
        throw e; 
      } catch (Exception e) { 
        throw new RuntimeException(e); 
      }
    }

    @Override 
    public void forEachRemaining(Consumer<? super String[]> action) {
      if (action == null) throw new NullPointerException();
      try {
        for (String[] row; (row = jr.readNext()) != null;) action.accept(row);
      } catch (RuntimeException e) {  
        throw e; 
      } catch (Exception e) { 
        throw new RuntimeException(e); 
      }
    }
}

这是我的主要 class，它读取输入文件并开始转换过程：-

import static java.util.concurrent.TimeUnit.SECONDS;

import java.io.IOException;
import java.io.PrintWriter;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.stream.Stream;

import com.dj.gls.core.FixedBatchSpliterator;


public class SpliteratorBenchmark {
    static double sink;

    public static void main(String[] args) throws IOException {
        final Path inputPath = createInput();

        for (int i = 0; i < 3; i++) {
            System.out.println("Start processing JDK stream");
            measureProcessing(Files.lines(inputPath));
            System.out.println("Start processing fixed-batch stream");
            measureProcessing(FixedBatchSpliterator.withBatchSize(Files.lines(inputPath), 10));
        }
    }

    private static void measureProcessing(Stream<String> input) throws IOException {
        final long start = System.nanoTime();
        try (Stream<String> lines = input) {
            final long totalTime = lines.parallel().mapToLong(SpliteratorBenchmark::processLine).sum();
            final double cpuTime = totalTime, realTime = System.nanoTime() - start;
            final int cores = Runtime.getRuntime().availableProcessors();
            System.out.println("          Cores: " + cores);
            System.out.format("       CPU time: %.2f s\n", cpuTime / SECONDS.toNanos(1));
            System.out.format("      Real time: %.2f s\n", realTime / SECONDS.toNanos(1));
            System.out.format("CPU utilization: %.2f%%\n\n", 100.0 * cpuTime / realTime / cores);
        }
    }


    private static long processLine(String line) {
        final long localStart = System.nanoTime();

        //conversion of json to xml and storing an xml file


        return System.nanoTime() - localStart;
    }

    private static Path createInput() throws IOException {
        final Path inputPath = Paths.get("/src/main/resources/input/input.txt");
        return inputPath;
    }
}

这就是我计划将 json 转换为 xml 的方式：-

import java.io.InputStream;
import java.io.OutputStream;
import java.io.Reader;
import java.io.Writer;
import javax.json.Json;
import javax.json.stream.JsonParser;
import javax.xml.stream.XMLOutputFactory;
import javax.xml.stream.XMLStreamException;
import javax.xml.stream.XMLStreamWriter;
import javax.xml.transform.Result;

/**
 * Converts JSON stream to XML stream.
 * Uses XML representation of JSON defined in XSLT 3.0.
 * 
 * @see <a href="https://www.w3.org/TR/xslt-30/#json">22 Processing JSON Data</a>
 * 
 */
public class JsonStreamXMLWriter
{

    public static final String XPATH_FUNCTIONS_NS = "http://www.w3.org/2005/xpath-functions";
    private static final XMLOutputFactory XOF = XMLOutputFactory.newInstance();

    static
    {
        XOF.setProperty(XMLOutputFactory.IS_REPAIRING_NAMESPACES, true);
    }

    private final JsonParser parser;
    private final XMLStreamWriter writer;

    public JsonStreamXMLWriter(Reader reader, Writer stream) throws XMLStreamException
    {
        this(Json.createParser(reader), getXMLOutputFactory().createXMLStreamWriter(stream));
    }

    public JsonStreamXMLWriter(Reader reader, OutputStream stream) throws XMLStreamException
    {
        this(Json.createParser(reader), getXMLOutputFactory().createXMLStreamWriter(stream));
    }

    public JsonStreamXMLWriter(Reader reader, OutputStream stream, String encoding) throws XMLStreamException
    {
        this(Json.createParser(reader), getXMLOutputFactory().createXMLStreamWriter(stream, encoding));
    }

    public JsonStreamXMLWriter(Reader reader, Result result) throws XMLStreamException
    {
        this(Json.createParser(reader), getXMLOutputFactory().createXMLStreamWriter(result));
    }

    public JsonStreamXMLWriter(Reader reader, XMLStreamWriter writer)
    {
        this(Json.createParser(reader), writer);
    }

    public JsonStreamXMLWriter(InputStream is, Writer stream) throws XMLStreamException
    {
        this(Json.createParser(is), getXMLOutputFactory().createXMLStreamWriter(stream));
    }

    public JsonStreamXMLWriter(InputStream is, OutputStream stream) throws XMLStreamException
    {
        this(Json.createParser(is), getXMLOutputFactory().createXMLStreamWriter(stream));
    }

    public JsonStreamXMLWriter(InputStream is, OutputStream stream, String encoding) throws XMLStreamException
    {
        this(Json.createParser(is), getXMLOutputFactory().createXMLStreamWriter(stream, encoding));
    }

    public JsonStreamXMLWriter(InputStream is, Result result) throws XMLStreamException
    {
        this(Json.createParser(is), getXMLOutputFactory().createXMLStreamWriter(result));
    }

    public JsonStreamXMLWriter(InputStream is, XMLStreamWriter writer)
    {
        this(Json.createParser(is), writer);
    }

    public JsonStreamXMLWriter(JsonParser parser, Writer stream) throws XMLStreamException
    {
        this(parser, getXMLOutputFactory().createXMLStreamWriter(stream));
    }

    public JsonStreamXMLWriter(JsonParser parser, OutputStream stream) throws XMLStreamException
    {
        this(parser, getXMLOutputFactory().createXMLStreamWriter(stream));
    }

    public JsonStreamXMLWriter(JsonParser parser, OutputStream stream, String encoding) throws XMLStreamException
    {
        this(parser, getXMLOutputFactory().createXMLStreamWriter(stream, encoding));
    }

    public JsonStreamXMLWriter(JsonParser parser, Result result) throws XMLStreamException
    {
        this(parser, getXMLOutputFactory().createXMLStreamWriter(result));
    }

    public JsonStreamXMLWriter(JsonParser parser, XMLStreamWriter writer)
    {
        this.parser = parser;
        this.writer = writer;
    }

    public void convert() throws XMLStreamException
    {
        convert(getWriter());
    }

    public void convert(XMLStreamWriter writer) throws XMLStreamException
    {
        convert(getParser(), writer);
    }

    public static void convert(JsonParser parser, XMLStreamWriter writer) throws XMLStreamException
    {
        writer.writeStartDocument();
        writer.setDefaultNamespace(XPATH_FUNCTIONS_NS);

        write(parser, writer);

        writer.writeEndDocument();
        writer.flush();

        parser.close();
    }

    public static void write(JsonParser parser, XMLStreamWriter writer) throws XMLStreamException
    {
        String keyName = null;
        while (parser.hasNext())
        {
            JsonParser.Event event = parser.next();

            switch(event)
            {
                case START_ARRAY:
                    writer.writeStartElement(XPATH_FUNCTIONS_NS, "array");
                    if (keyName != null)
                    {
                        writer.writeAttribute("key", keyName);
                        keyName = null;
                    }
                break;
                case END_ARRAY:
                    writer.writeEndElement();
                break;
                case START_OBJECT:
                    writer.writeStartElement(XPATH_FUNCTIONS_NS, "map");
                    if (keyName != null)
                    {
                        writer.writeAttribute("key", keyName);
                        keyName = null;
                    }
                break;
                case END_OBJECT:
                    writer.writeEndElement();
                break;
                case VALUE_FALSE:
                    writer.writeStartElement(XPATH_FUNCTIONS_NS, "boolean");
                    if (keyName != null)
                    {
                        writer.writeAttribute("key", keyName);
                        keyName = null;
                    }
                    writer.writeCharacters("false");
                    writer.writeEndElement();
                break;
                case VALUE_TRUE:
                    writer.writeStartElement(XPATH_FUNCTIONS_NS, "boolean");
                    if (keyName != null)
                    {
                        writer.writeAttribute("key", keyName);
                        keyName = null;
                    }
                    writer.writeCharacters("true");
                    writer.writeEndElement();
                break;
                case KEY_NAME:
                    keyName = parser.getString();
                break;
                case VALUE_STRING:
                    writer.writeStartElement(XPATH_FUNCTIONS_NS, "string");
                    if (keyName != null)
                    {
                        writer.writeAttribute("key", keyName);
                        keyName = null;
                    }
                    writer.writeCharacters(parser.getString());
                    writer.writeEndElement();
                break;
                case VALUE_NUMBER:
                    writer.writeStartElement(XPATH_FUNCTIONS_NS, "number");
                    if (keyName != null)
                    {
                        writer.writeAttribute("key", keyName);
                        keyName = null;
                    }
                    writer.writeCharacters(parser.getString());
                    writer.writeEndElement();
                break;
                case VALUE_NULL:
                    writer.writeEmptyElement(XPATH_FUNCTIONS_NS, "null");
                    if (keyName != null)
                    {
                        writer.writeAttribute("key", keyName);
                        keyName = null;
                    }
                break;
            }

            writer.flush();
        }
    }

    protected JsonParser getParser()
    {
        return parser;
    }

    protected XMLStreamWriter getWriter()
    {
        return writer;
    }

    protected static XMLOutputFactory getXMLOutputFactory()
    {
        return XOF;
    }

}

这是我的class调用这个Utilclass来转换:-

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Reader;
import java.nio.charset.StandardCharsets;
import javax.xml.stream.XMLStreamException;

import org.apache.commons.io.IOUtils;


public class JSON2XML
{

    public static void main(String[] args) throws IOException, XMLStreamException
    {
        OutputStreamWriter outputStreamWriter = new OutputStreamWriter(outputStream);

        InputStream json = new FileInputStream(file);

        if (json.available() == 0)
        {
            System.err.println("JSON input: stdin");
            System.exit(1);
        }

        try (Reader reader = new BufferedReader(new InputStreamReader(json, StandardCharsets.UTF_8)))
        {

            new JsonStreamXMLWriter(reader, new BufferedWriter(new OutputStreamWriter(outputStream))).convert();
        }
    }

}

Answer 1

您可以做的最好的事情是使用分析器查看 CPU 时间和内存在哪里花费了大部分可用资源，并根据此进行优化。

Answer 2

我能够一个接一个地读取多个 json 文件并创建新的 xml 文件。下面是代码：-

private static long processLine(Path line) {
        final long localStart = System.nanoTime();
        System.out.println("Path" + line);
        OutputStream outputStream;
        try {
            if(Files.isRegularFile(line)) {
                String str = line.getFileName().toString();
                System.out.println("FIlename - " + str.substring(0, str.length()-4));

                outputStream = new FileOutputStream("/Users/d0s04ub/Documents/STS_Workspace/OrderProcessor/src/main/resources/output/" + str.substring(0, str.length()-4));

                InputStream content = Files.newInputStream(line);

                try (Reader reader = new BufferedReader(new InputStreamReader(content, StandardCharsets.UTF_8)))
                {
                    try {
                        new JsonStreamXMLWriter(reader, new BufferedWriter(new OutputStreamWriter(outputStream))).convert();
                    } catch (XMLStreamException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException e1) {
            e1.printStackTrace();
        }finally{

        }


        return System.nanoTime() - localStart;
    }

java 中多个 json 到 xml 的高效处理

efficient processing of multiple json to xml in java

java

json

java-stream

spliterator