本文主要是介绍响应式异步编程库Reactor的使用
响应式流简介
When the publisher is faster than the subscriber, the latter must have an unbounded buffer to store fast incoming items or it must drop items it cannot handle.Another solution is to use a strategy called backpressure in which the subscriber tells the publisher to slow down and hold the tems until the subscriber is ready to process more. Using backpressure may require the publisher to have an unbounded buffer if it keeps producing and storing elements for slower subscribers.The publisher may implement a bounded buffer to store a limited number of elements and may choose to drop them if its buffer is full.
What does the subscriber do when it requests items from the publisher and the items are not available?In a synchronous request, the subscriber must wait, possibly indefinitely, until items are available. If the publisher sends items to the subscriber synchronously and the subscriber processes them synchronously, the publisher must block until the data processing finishes. The solution is to have an asynchronous processing at both ends, where the subscriber may keep working on other tasks after requesting items from the publisher. When more items are ready, the publisher sends them to the subscriber asynchronously.
Reactive Streams started in 2013 as an initiative for providing a standard for asynchronous stream processing with non-blocking backpressure. It is aimed at solving the problems of processing a stream of items—how do you pass a stream of items from a publisher to a subscriber without requiring the publisher to block or the subscriber to have an unbounded buffer or drop.
The Reactive Streams model is very simple—the subscriber sends an asynchronous request to the publisher for N items. The publisher sends N or fewer items to the subscriber asynchronously.
Reactive Streams dynamically switches between the pull model and the push model streamprocessing mechanisms. It uses the pull model when the subscriber is slower and uses the push model when the subscriber is faster.
Reactor介绍
webflux与webmvc的类比:
| webmvc | webflux |
|---|---|
| controller | handler |
| request mapping | router |
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.1.4.RELEASE</version>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<version>3.1.4.RELEASE</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
<scope>test</scope>
</dependency>
一，Flue和Mono的简单用法
import java.util.Arrays;
import java.util.List;
import java.util.stream.Stream;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
public class SimpleReactor
{
private static void testConstructUsingJust()
{
//subscribe方法中的lambda表达式作用在了每一个数据元素上
Flux.just(1, 2, 3, 4, 5, 6).subscribe(System.out::print);
System.out.println();//回车的作用
Mono.just(1).subscribe(System.out::println);
}
private static void testConstructFromArray()
{
Integer[] array = new Integer[]{1,2,3,4,5,6};
Flux.fromArray(array).subscribe(x -> {
System.out.println("收到 "+ x);
});
}
private static void testConstructFromList()
{
List<Integer> list = Arrays.asList(1,2,3,4,5,6);
Flux<Integer> flux = Flux.fromIterable(list);
flux.subscribe(
System.out::println,
System.err::println,
() -> System.out.println("Completed!"));
}
private static void testConstructFromStream()
{
Stream<Integer> stream = Arrays.asList(1,2,3,4,5,6).stream();
Flux.fromStream(stream).subscribe(System.out::print);
}
private static void testMonoError()
{
Mono.error(new Exception(" 注意注意，发生异常，注意处理啦")).subscribe(
System.out::println,
System.err::println,
() -> System.out.println("Completed!")
);
}
public static void main( String[] args )
{
testConstructUsingJust();
nextTest();
testConstructFromArray();
nextTest();
testConstructFromList();
nextTest();
testMonoError();
nextTest();
testConstructFromStream();
}
private static void nextTest()
{
System.out.println("********************************");
}
}
二、Reactor中如何做單元測試
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;
public class SimpleReactorTest
{
private Flux<Integer> generateFluxFrom1To6()
{
return Flux.just(1, 2, 3, 4, 5, 6);
}
private Mono<Integer> generateMonoWithError()
{
return Mono.error(new Exception("some error"));
}
@Test
public void testViaStepVerifier()
{
StepVerifier.create(generateFluxFrom1To6()).expectNext(1, 2, 3, 4, 5, 6)
.expectComplete().verify();
StepVerifier.create(generateMonoWithError())
.expectErrorMessage("some error").verify();
}
}
三、Flux和Mono也支持map、flatMap、Filter、zip等operator
@Test
public void testMapAndFlatMap()
{
// 注意下面的6表示6個,和IntStream的Range方法里面不一样
StepVerifier.create(Flux.range(1, 6).map(i -> i * i))
.expectNext(1, 4, 9, 16, 25, 36).expectComplete();
StepVerifier
.create(Flux.just("flux", "mono")
.flatMap(s -> Flux.fromArray(s.split("\\s*"))
.delayElements(Duration.ofMillis(100)))
.doOnNext(System.out::print))
.expectNextCount(8).verifyComplete();
}
@Test
public void testFilter()
{
StepVerifier.create(Flux.range(1, 6).filter(i -> i % 2 == 1) // 1
.map(i -> i * i)).expectNext(1, 9, 25) // 2
.verifyComplete();
}
private Flux<String> getZipDescFlux()
{
String desc = "Zip two sources together, that is to say wait for all the sources to emit one element and combine these elements once into a Tuple2.";
return Flux.fromArray(desc.split("\\s+")); // 1
}
@Test
public void testZip() throws InterruptedException
{
CountDownLatch countDownLatch = new CountDownLatch(1);
//使用Flux.interval声明一个每200ms发出一个元素的long数据流；因为zip操作是一对一的，故而将其与字符串流zip之后，字符串流也将具有同样的速度；
Flux.zip(getZipDescFlux(), Flux.interval(Duration.ofMillis(200)))
.subscribe(
t -> System.out.println(t.getT1()),
null,
countDownLatch::countDown); // 4
countDownLatch.await(10, TimeUnit.SECONDS); // 5
}
有些内容还没有研究完，请接着看 http://blog.51cto.com/liukang/2090191或者https://www.ibm.com/developerworks/cn/java/j-cn-with-reactor-response-encode/index.html
