【发布时间】：2020-01-10 22:42:25
【问题描述】：
我创建了一个 Flux，它不断产生一个新的整数值。我有一个订阅者在自己的线程中运行（.publishOn(single())）。无论我使用什么策略（最新或其他），我总是得到相同的结果：
*** Received 1 with thread single-1
>>> Generated 1 with thread main
>>> Generated 2 with thread main
>>> Generated 3 with thread main
>>> Generated 4 with thread main
>>> Generated 5 with thread main
>>> Generated 6 with thread main
>>> Generated 7 with thread main
>>> Generated 8 with thread main
>>> Generated 9 with thread main
*** Received 2 with thread single-1
*** Received 3 with thread single-1
*** Received 4 with thread single-1
*** Received 5 with thread single-1
*** Received 6 with thread single-1
*** Received 7 with thread single-1
据我了解，通过设置latest，我应该只收到最后一个整数.. 应该删除一些整数？
@Test
@DisplayName("test")
public void workingFlux() throws InterruptedException {
CountDownLatch latch = new CountDownLatch(1);
Flux<Integer> IntGenerator = Flux.create(e -> {
AtomicInteger iteration = new AtomicInteger(1);
while (iteration.intValue() < 10) {
int value = iteration.getAndIncrement();
e.next(value);
if (value < 10) {
System.out.println(">>> Generated " + value + " with thread " + Thread.currentThread().getName());
}
}
}, FluxSink.OverflowStrategy.DROP);
IntGenerator.publishOn(single())
.subscribe(new Subscriber<Integer>() {
private Subscription s;
@Override
public void onSubscribe(final Subscription subscription) {
s = subscription;
s.request(1);
}
@Override
public void onNext(final Integer integer) {
System.out.println("*** Received " + integer + " with thread " + Thread.currentThread().getName());
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
s.request(1);
}
@Override
public void onError(final Throwable throwable) {
}
@Override
public void onComplete() {
}
});
Flux<Integer> IntGenerator = Flux.create(e -> {
AtomicInteger iteration = new AtomicInteger(1);
while (iteration.intValue() < 10) {
int value = iteration.getAndIncrement();
e.next(value);
if (value < 10) {
System.out.println(">>> Generated " + value + " with thread " + Thread.currentThread().getName());
}
}
}, FluxSink.OverflowStrategy.LATEST);
IntGenerator.publishOn(single())
.subscribe(new Subscriber<Integer>() {
private Subscription s;
@Override
public void onSubscribe(final Subscription subscription) {
System.out.println("Subscribed");
s = subscription;
s.request(1);
}
@Override
public void onNext(final Integer integer) {
System.out.println("*** Received " + integer + " with thread " + Thread.currentThread().getName());
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
s.request(1);
}
@Override
public void onError(final Throwable throwable) {
}
@Override
public void onComplete() {
}
});
latch.await(120L, TimeUnit.SECONDS);
latch.await(120L, TimeUnit.SECONDS);
}
【问题讨论】：
-
您有什么问题？和/或您期望的结果是什么？请edit您的问题将您拥有的源代码包含为minimal reproducible example，其他人可以编译和测试。