Substituting 'synchronized' with faster and fully thread-safe collections 'ConcurrentLinkedDeque' and iterators.
import java.io.File;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
+import java.util.Deque;
import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.Deflater;
-import static java.util.Collections.synchronizedList;
import static org.apache.commons.compress.archivers.zip.ZipArchiveEntryRequest.createZipArchiveEntryRequest;
/**
* @since 1.10
*/
public class ParallelScatterZipCreator {
- private final List<ScatterZipOutputStream> streams = synchronizedList(new ArrayList<ScatterZipOutputStream>());
+ private final Deque<ScatterZipOutputStream> streams = new ConcurrentLinkedDeque<>();
private final ExecutorService es;
private final ScatterGatherBackingStoreSupplier backingStoreSupplier;
- private final List<Future<ScatterZipOutputStream>> futures = new ArrayList<>();
+ private final Deque<Future<ScatterZipOutputStream>> futures = new ConcurrentLinkedDeque<>();
private final long startedAt = System.currentTimeMillis();
private long compressionDoneAt = 0;
@@ -256,8 +255,6 @@ public void writeTo(final ZipArchiveOutputStream targetStream)
// It is important that all threads terminate before we go on, ensure happens-before relationship
compressionDoneAt = System.currentTimeMillis();
- synchronized (streams) {
- // write zip entries in the order they were added (kept as futures)
for (final Future<ScatterZipOutputStream> future : futures) {
ScatterZipOutputStream scatterStream = future.get();
scatterStream.zipEntryWriter().writeNextZipEntry(targetStream);
@@ -266,7 +263,6 @@ public void writeTo(final ZipArchiveOutputStream targetStream)
for (final ScatterZipOutputStream scatterStream : streams) {
scatterStream.close();
}
- }
scatterDoneAt = System.currentTimeMillis();
} finally {
@@ -284,7 +280,6 @@ public ScatterStatistics getStatisticsMessage() {
}
private void closeAll() {
- synchronized (streams) {
for (final ScatterZipOutputStream scatterStream : streams) {
try {
scatterStream.close();
@@ -294,5 +289,4 @@ private void closeAll() {
}
}
}
-}
