【发布时间】：2020-11-30 15:50:47
【问题描述】：
以下多线程示例从两个不同的线程写入TopicProcessor，并在两个不同的线程中从TopicProcessor 读取。但是，某处存在竞争条件，因此并非所有事件都传递给订阅者，导致应用程序在processed.await() 中永远挂起。有人知道为什么吗？
import reactor.core.publisher.Flux;
import reactor.extra.processor.TopicProcessor;
import reactor.extra.processor.WaitStrategy;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static java.util.Arrays.asList;
public class ReactorTopicProcessorSample {
public static class Producer implements Callable<Void> {
final String name;
final List<String> data;
final CountDownLatch producerCount;
final TopicProcessor<String> topicProcessor;
public Producer(String name, List<String> data, CountDownLatch submitted, TopicProcessor<String> topicProcessor) {
this.name = name;
this.data = data;
this.producerCount = submitted;
this.topicProcessor = topicProcessor;
}
@Override
public Void call() throws Exception {
producerCount.countDown();
producerCount.await(); // wait until the other producer is submitted to be sure that they run in different threads
Flux.fromIterable(data)
.map(s -> "\"" + s + "\"" + " from thread " + Thread.currentThread().getName())
.delayElements(Duration.ofMillis(10))
.subscribe(topicProcessor);
System.out.println("Submitted " + name + " producer in thread " + Thread.currentThread().getName() + ".");
return null;
}
}
public static void main(String[] args) throws InterruptedException {
for (int i = 0; i < 100; i++) { // this sample doesn't hang every time. repeat a few times to make it reproducible
realMain(args);
System.out.println("\n--- the previous run was successful. running again ---\n");
}
}
public static void realMain(String[] args) throws InterruptedException {
List<String> numbers = asList("1", "2", "3", "4", "5", "6", "7", "8");
List<String> characters = asList("a", "b", "c", "d", "e", "f", "g", "h");
CountDownLatch producerCount = new CountDownLatch(2);
CountDownLatch subscriberCount = new CountDownLatch(2);
CountDownLatch processed = new CountDownLatch(
(int) subscriberCount.getCount() * (numbers.size() + characters.size()));
ExecutorService exec = Executors.newFixedThreadPool((int) producerCount.getCount());
TopicProcessor<String> topicProcessor = TopicProcessor.<String>builder()
.share(true)
.name("topic-processor")
.bufferSize(16)
.waitStrategy(WaitStrategy.liteBlocking())
.build();
Flux<String> flux = Flux.from(topicProcessor)
.doOnSubscribe(s -> subscriberCount.countDown());
flux.subscribe(out -> {
System.out.println("Subscriber in thread " + Thread.currentThread().getName() + " received " + out);
processed.countDown();
});
flux.subscribe(out -> {
System.out.println("Subscriber in thread " + Thread.currentThread().getName() + " received " + out);
processed.countDown();
});
subscriberCount.await();
exec.submit(new Producer("number", numbers, producerCount, topicProcessor));
exec.submit(new Producer("character", characters, producerCount, topicProcessor));
processed.await();
exec.shutdown();
topicProcessor.shutdown();
}
}
依赖关系
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.3.2.RELEASE</version>
</dependency>
<dependency>
<groupId>io.projectreactor.addons</groupId>
<artifactId>reactor-extra</artifactId>
<version>3.3.2.RELEASE</version>
</dependency>
示例行为：订阅者只接收字符或只接收数字，导致程序在processed.await() 中永远等待。这并非每次都会发生，有时会按预期工作。
【问题讨论】：
标签： java project-reactor