用java-8流混合显式和隐式并行性

Question

在过去，我写过一些java程序，使用两个线程。第一个线程（生产者）从API（C库）读取数据，创建一个java对象，将该对象发送到另一个线程。C API正在传递一个事件流（无限）。线程使用LinkedBlockingQueue作为管道交换对象(put、poll)。第二个线程（使用者）正在处理对象。（我还发现，线程中的代码更易读。第一个线程处理C API的内容并生成适当的java对象，第二个线程不受C API处理的影响，而是处理数据）。

现在我感兴趣的是如何使用Java8中的新流API实现上面的场景。但是假设我想保留两个线程（生产者/消费者）！第一个线程正在写入流。第二个线程正在从流中读取。我还希望，我可以使用这种技术处理更好的显式并行（生产者/消费者），并在流中使用一些隐式并行（例如，stream.parallel（））。

我对新的流API没有太多的经验。因此，我尝试了下面的代码，以解决上面的想法。

null

null

再说一次：这个想法是与线程有一个显式的并行性。但在内部，我可以处理新特性，并在可能的情况下使用并行处理

也感谢你对这个问题的关心。

package sandbox.test;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.LongStream;

public class MyStream {
    private volatile LongStream stream = null;
    private AtomicInteger producerCount = new AtomicInteger(0);
    private AtomicInteger consumerCount = new AtomicInteger(0);
    private AtomicInteger apiError = new AtomicInteger(0);

    public static void main(String[] args) throws InterruptedException {
    MyStream appl = new MyStream();
    appl.create();
    }

    private static void sleep(long sleep) {
    try {
        Thread.sleep(sleep);
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
    }

    private static void apiError(final String pos, final int iteration) {
    RuntimeException apiException = new RuntimeException("API error pos=" + pos + " iteration=" + iteration);
    System.out.println(apiException.getMessage());
    throw apiException;
    }

    final private int simulateErrorAfter = 10;

    private Thread produce() {
    Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
        System.out.println("Producer started");
        stream = LongStream.generate(() -> {
            int localCount;
            // Detect error, while using stream.parallel() processing
            int error = apiError.get();
            if ( error > 0 )
                apiError("1", error);
            // ----- Accessing the C API here -----
            localCount = producerCount.incrementAndGet(); // C API access; delegate for accessing the C API
            // ----- Accessing the C API here -----

            // Checking error code from C API
            if ( localCount > simulateErrorAfter ) { // Simulate an API error
                producerCount.decrementAndGet();
                stream.close();
                apiError("2", apiError.incrementAndGet());
            }
            System.out.println("P: " + localCount);
            sleep(200L);
            return localCount;
            });
        System.out.println("Producer terminated");
        }
    });
    thread.start();
    return thread;
    }

    private Thread consume() {
    Thread thread = new Thread(new Runnable() {
        @Override
        public void run() {
        try {
            stream.onClose(new Runnable() {
            @Override
            public void run() {
                System.out.println("Close detected");
            }
            }).parallel().forEach(l -> {
            sleep(1000);
            System.out.println("C: " + l);
            consumerCount.incrementAndGet();
            });
        } catch (Exception e) {
            // Capturing the stream end
            System.out.println(e);
        }
        System.out.println("Consumer terminated");
        }
    });
    thread.start();
    return thread;
    }

    private void create() throws InterruptedException {
    Thread producer = produce();
    while ( stream == null )
        sleep(10);
    Thread consumer = consume();
    producer.join();
    consumer.join();
    System.out.println("Produced: " + producerCount);
    System.out.println("Consumed: " + consumerCount);

    }
}

Answer 1

您需要了解流API的一些基本要点：

>

在流上应用的所有操作都是惰性的，在应用终端操作之前不会执行任何操作。使用“生产者”线程创建流是没有意义的，因为这个线程不会做任何事情。所有操作都在您的“使用者”线程和由流实现本身启动的后台线程中执行。创建流实例的线程完全不相关

关闭流与stream操作本身无关，即不会关闭线程。它意味着释放额外的资源，例如关闭与files.lines(…)返回的流相关联的文件。您可以使用onclose调度这样的清理操作，当您调用close时，流将调用这些操作，但仅此而已。对于Stream类本身来说，它没有任何意义。