Java 9-发布者和订阅者的工作方式

Question

我正在努力了解Subscriber和Publisher在java 9中是如何工作的。

在这里，我在这里创建了一个subscriber，并使用SubmissionPublisher发布项目。

我正在尝试将100个字符串发布到subscriber。如果我不将Client程序设置为sleep(请参阅MyReactiveApp中的注释代码(，则不会看到所有项目都已发布。

为什么它不等待这里处理的所有字符串：

strs.stream().forEach(i -> publisher.submit(i)); // what happens here? 

如果我用替换上面的代码，我会看到所有字符串都打印在控制台中

strs.stream().forEach(System.out::println);

使用SubmissionPublisher发布的客户端程序。

import java.util.List;
import java.util.concurrent.SubmissionPublisher;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
public class MyReactiveApp {
public static void main(String args[]) throws InterruptedException {
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
MySubscriber subs = new MySubscriber();
publisher.subscribe(subs);

List<String> strs = getStrs();
System.out.println("Publishing Items to Subscriber");
strs.stream().forEach(i -> publisher.submit(i));
/*while (strs.size() != subs.getCounter()) {
Thread.sleep(10);
}*/
//publisher.close();
System.out.println("Exiting the app");
}
private static List<String> getStrs(){
return Stream.generate(new Supplier<String>() {
int i =1;
@Override
public String get() {
return "name "+ (i++);
}
}).limit(100).collect(Collectors.toList());
}
}

用户

import java.util.concurrent.Flow.Subscription;
public class MySubscriber implements java.util.concurrent.Flow.Subscriber<String>{
private Subscription subscription;
private int counter = 0;
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(100);
}
@Override
public void onNext(String item) {
System.out.println(this.getClass().getSimpleName()+" item "+item);
//subscription.request(1);
counter++;
}
@Override
public void onError(Throwable throwable) {
System.out.println(this.getClass().getName()+ " an error occured "+throwable);
}
@Override
public void onComplete() {
System.out.println("activity completed");
}
public int getCounter() {
return counter;
}
}

输出：

Publishing Items to Subscriber
MySubscriber item name 1
MySubscriber item name 2
MySubscriber item name 3
MySubscriber item name 4
MySubscriber item name 5
Exiting the app
MySubscriber item name 6
MySubscriber item name 7
MySubscriber item name 8
MySubscriber item name 9
MySubscriber item name 10
MySubscriber item name 11
MySubscriber item name 12

Answer 1

SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

使用ForkJoinPool.commonPool((创建一个新的SubmissionPublisher，用于异步传递到订阅者

请参阅：https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/SubmissionPublisher.html#SubmissionPublisher--

所以实际上

strs.stream().forEach(i -> publisher.submit(i));

将所有提交排入队列，并在另一个线程上异步传递它们。但随后应用程序被终止。这与工作线程的进度无关。这意味着无论工作线程已经交付了多少元素，应用程序都将终止。

每次跑步的情况可能不同。在最坏的情况下，应用程序可能会在第一个项目交付之前终止。

线程

如果你想验证MyReactiveApp的主要方法和MySubscriber的onNext中的交付是否发生在不同的线程上，你可以打印出相应线程的名称，例如在MyReactiveApp's main:中

System.out.println(Thread.currentThread().getName()) 

将输出CCD_ 10作为线程名称。

而MySubscriber的onNext方法将例如输出类似ForkJoinPool.commonPool-worker-1的内容。

用户和精灵线程

尽管我们还有一个正在运行的线程，为什么应用程序会终止？

Java中有两种线程：

• 用户线程
• 后台进程线程

当不再有任何用户线程在运行时，即使deamon线程仍在运行，Java程序也会终止。

主线程是一个用户线程。SubmissionPublisher在此处使用ForkJoinPool.commonPool((中的工作线程。这些是守护进程线程。

所有工作线程都使用Thread.isDemon((设置为true进行初始化。

https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/ForkJoinPool.html