本文主要是介紹響應(yīng)式異步編程庫Reactor的使用
響應(yīng)式流簡介
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>
一，F(xiàn)lue和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表達式作用在了每一個數(shù)據(jù)元素上
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(" 注意注意，發(fā)生異常，注意處理啦")).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發(fā)出一個元素的long數(shù)據(jù)流；因為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
}
有些內(nèi)容還沒有研究完，請接著看 http://blog.51cto.com/liukang/2090191或者https://www.ibm.com/developerworks/cn/java/j-cn-with-reactor-response-encode/index.html
