[JAVA] Informationen zur Implementierung der Zusammenführungsverarbeitung einschließlich der Sortierfunktion der Stream-API

Einführung

Es wurde beschlossen, zwei sortierte Dateien gemäß einer Anforderung für kleine Unternehmen zusammenzuführen und auszugeben.

Hier ist ein Beispiel

1. Die Eingabedatei besteht aus 2 Dateien, die nach dem ersten Schlüssel sortiert sind.
2. Die Ausgabedatei wird sortiert und aus allen Daten in den beiden Dateien ausgegeben.

** Eingabedatei 1 **

1,AAA
4,BBB
9,CCC

** Eingabedatei 2 **

2,DDD
5,EEE
6,FFF

** Ausgabedatei **

1,AAA
2,DDD
4,BBB
5,EEE
6,FFF
9,CCC

Deshalb Ich wollte mit der Strem-API lesen und die Zusammenführungssortierung aktivieren.

Beeindruckend, Es ist auch ein Thema, das zu genau ist und wenig Nachfrage besteht. (´ ・ ω ・)

Was Sie implementieren möchten

Ich habe beschlossen, es so umzusetzen.

1. Erstellen Sie einen einzelnen Stream mit mehreren Streams als Parametern.
2. Übergeben Sie Comparator, um die Sortierreihenfolge zu definieren.
3. Als Voraussetzung wird jeder Stream sortiert.
4. Staatenlos ist eine wichtige Voraussetzung.

Implementierung

Erstellen Sie StreamUtils, um Streams zusammenzuführen.

public class StreamUtils {

	private StreamUtils() {
	}

	/**
	 *Zusammenführen sortiert den durch das Argument angegebenen Stream
	 *
	 * @param streamArray Sortierter Stream
	 * @return
	 */
	@SafeVarargs
	public static final <T> Stream<T> merge(final Stream<T>... streamArray) {
		return merge(null, streamArray);
	}

	/**
	 *Streams zusammenführen
	 *Steuern Sie die erfassten Elemente mit Ihrem eigenen Spliterator.
	 *
	 * @param comp sort bedingung
	 * @param streamArray Sortierter Stream
	 * @return
	 */
	@SafeVarargs
	public static final <T> Stream<T> merge(final Comparator<T> comp, final Stream<T>... streamArray) {
		try {
			MergedIterator<T> iterator = new MergedIterator<>(streamArray);
			iterator.setComparetor(comp);
			Spliterator<T> spliterator = new SpliteratorAdapter<>(iterator);
			return StreamSupport.stream(spliterator, false).onClose(composedClose(streamArray));
		} catch (Exception exception) {
			for (Stream<?> stream : streamArray) {
				try {
					stream.close();
				} catch (RuntimeException e) {
					try {
						exception.addSuppressed(e);
					} catch (RuntimeException ignore) {
					}
				}
			}
			throw exception;
		}
	}

	@SafeVarargs
	static <T> Runnable composedClose(final Stream<T>... streamArray) {
		return new Runnable() {
			@Override
			public void run() {
				RuntimeException exception = null;
				for (Stream<?> stream : streamArray) {
					try {
						stream.close();
					} catch (RuntimeException e) {
						try {
							if (exception == null) {
								exception = e;
							} else {
								exception.addSuppressed(e);
							}
						} catch (RuntimeException ignore) {
						}
					}
				}
				if (exception != null) {
					throw exception;
				}
			}
		};
	}

}

Vergleichen Sie die Klasse, die mehrere Streams enthält, und wählen Sie den zu erfassenden Inhalt aus

public class MergedIterator<T> implements Iterator<T> {

	/**
	 *Karte des nächsten Elements der Strema-Instanz, die Sie halten
	 */
	private Map<Iterator<T>, T> nextMap;

	/**
	 *Sortierbedingungen
	 */
	private Comparator<T> comp = null;

	/**
	 * 
	 * @param streamArray Array von Streams, die zusammengeführt werden sollen
	 */
	@SafeVarargs
	public MergedIterator(final Stream<T>... streamArray) {
		this(Arrays.asList(streamArray).stream().map(stream -> stream.iterator()).collect(Collectors.toList()));
	}

	/**
	 * 
	 * @param itrArray Array von Iteratoren, die zusammengeführt werden sollen
	 */
	@SafeVarargs
	public MergedIterator(final Iterator<T>... itrArray) {
		this(Arrays.asList(itrArray));
	}

	/**
	 *
	 * @param itrList Liste der zusammenzuführenden Iteratoren
	 */
	public MergedIterator(final List<Iterator<T>> itrList) {
		this.nextMap = new HashMap<>();
		for (Iterator<T> itr : itrList) {
			this.nextMap.put(itr, itr.hasNext() ? itr.next() : null);
		}
	}

	/**
	 *Instanz zum Vergleich
	 * @param comp
	 */
	public void setComparetor(final Comparator<T> comp) {
		this.comp = comp;
	}

	@Override
	public boolean hasNext() {
		return this.nextMap.entrySet().stream().filter(entry -> entry.getValue() != null).count() > 0L;
	}

	@Override
	public void remove() {
		throw new UnsupportedOperationException("Not supported.");
	}

	@Override
	public T next() {
		if (!hasNext()) {
			return null;
		}

		Entry<Iterator<T>, T> nextEntry = this.nextMap.entrySet().stream().filter(entry -> entry.getValue() != null)
				.min(new Comparator<Entry<Iterator<T>, T>>() {

					@Override
					public int compare(final Entry<Iterator<T>, T> o1, final Entry<Iterator<T>, T> o2) {
						return MergedIterator.this.comp != null
								? MergedIterator.this.comp.compare(o1.getValue(), o2.getValue())
								: 0;
					}
				}).orElse(null);

		T returnObject = nextEntry.getValue();
		nextEntry.setValue(nextEntry.getKey().hasNext() ? nextEntry.getKey().next() : null);

		return returnObject;

	}

}

Spliterator-Klasse (Element-Scanning innerhalb von Stream)

public class SpliteratorAdapter<T> extends Spliterators.AbstractSpliterator<T> {

	private final Iterator<T> iterator;

	/**
	 *
	 * @param iter
	 */
	public SpliteratorAdapter(final Iterator<T> iter) {
		super(Long.MAX_VALUE, 0);
		this.iterator = iter;
	}

	@Override
	public synchronized boolean tryAdvance(final Consumer<? super T> action) {
		if (this.iterator.hasNext()) {
			action.accept(this.iterator.next());
			return true;
		}
		return false;
	}
}

Ausführungscode

Nennen Sie es wie folgt. Stream-Ressourcen werden nach dem Zusammenführen geschlossen, müssen jedoch einzeln erstellt werden (streamArray). Ich frage mich, ob ich es klug machen kann. .. .. ..

Es kann einfacher sein, wenn Sie dies in StreamUtils aufnehmen.

	//streamArray Strema-Array
	//Implementierung von comp Comparator
	try (Stream<List<String>> mergeStream = StreamUtils
				.merge(comp, streamArray)
				.onClose(() -> LOG.debug("Alle Stream-Verarbeitung abgeschlossen"))) {
		/*Hier wird die Verarbeitung ausgeführt. In diesem Beispiel wird die Standardausgabe ausgegeben und der Prozess beendet.*/
		mergeStream.forEach(line -> System.out.println(line));

	} catch (final IOException e) {
		//Ausnahmebehandlung
	}

abschließend

Ich denke, dass es einige Prozesse gibt, die mehrere Streams umfassen, außer sie in Reihe zu halten. DIFF kann also auch zustandslos ausgeführt werden Ich denke, es wäre interessant, diesen Bereich zu schaffen.

In diesem Beispiel funktioniert die parallele Option nicht. Ich denke, es gibt eine Verarbeitung, die funktioniert.

Ich würde es gerne einmal organisieren.