87

Reactor of Java 这一章来自于《Spring in Action, 5th》 的笔记，因为这本书讲Reactor of Java讲的太好了，所以作为笔记摘抄了下来。

Reactor of Java
In an imperative programming model, the code would look something like this:

String name = “Craig”;
String capitalName = name.toUpperCase();
String greeting = "Hello, " + capitalName + “!”;
System.out.println(greeting);
In the imperative model, each line of code performs a step, one right after the other, and definitely in the same thread. Each step blocks the executing thread from moving to the next step until complete. In contrast, functional, reactive code could achieve the same thing like this:

Mono.just(“Craig”)
.map(n -> n.toUpperCase())
.map(n -> “Hello, " + n + " !”)
.subscribe(System.out::println);
The Mono in the example is one of Reactor’s two core types. Flux is the other. Both are implementations of Reactive Streams’ Publisher.
A Flux represents** a pipeline of zero, one, or many (potentially infinite) data items**.
A Mono is a specialized reactive type that’s optimized for when the dataset is known to have no more than one data item.

CREATING FROM OBJECTS

Flux fruitFlux = Flux
.just(“Apple”, “Orange”, “Grape”, “Banana”, “Strawberry”);
fruitFlux.subscribe(f -> System.out.println("Hello " + f));

// for test
StepVerifier.create(fruitFlux)
.expectNext(“Apple”)
.expectNext(“Orange”)
.expectNext(“Grape”)
.expectNext(“Banana”)
.expectNext(“Strawberry”)
.verifyComplete();
CREATING FROM COLLECTIONS

Stream fruitStream = Stream.of(“Apple”, “Orange”, “Grape”, “Banana”, “Strawberry”);
Flux fruitFlux2 = Flux.fromStream(fruitStream);
fruitFlux2.subscribe(s -> System.out.println(s));

    List<String> fruitList = new ArrayList<>();
    fruitList.add("Apple");
    fruitList.add("Orange");
    fruitList.add("Grape");
    fruitList.add("Banana");
    fruitList.add("Strawberry");
    Flux<String> fruitFlux3 = Flux.fromIterable(fruitList);
    fruitFlux3.subscribe(s -> System.out.println(s));


    String[] fruits = new String[] {"Apple", "Orange", "Grape", "Banana", "Strawberry" };
    Flux<String> fruitFlux = Flux.fromArray(fruits);
    fruitFlux.subscribe(s -> System.out.println(s));
    StepVerifier.create(fruitFlux)
    .expectNext("Apple")
    .expectNext("Orange")
    .expectNext("Grape")
    .expectNext("Banana")
    .expectNext("Strawberry")
    .verifyComplete();

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19
• 20

GENERATING FLUX DATA

Flux intervalFlux =
Flux.range(1, 5);
intervalFlux.subscribe(integer -> System.out.println(integer));
StepVerifier.create(intervalFlux)
.expectNext(1)
.expectNext(2)
.expectNext(3)
.expectNext(4)
.expectNext(5)
.verifyComplete();

Flux intervalFlux =
Flux.interval(Duration.ofSeconds(1))
.take(5);
intervalFlux.subscribe(i -> System.out.println(i));
StepVerifier.create(intervalFlux)
.expectNext(0L)
.expectNext(1L)
.expectNext(2L)
.expectNext(3L)
.expectNext(4L)
.verifyComplete();
MERGING REACTIVE TYPES

Flux characterFlux = Flux
.just(“Garfield”, “Kojak”, “Barbossa”)
.delayElements(Duration.ofMillis(500));
Flux foodFlux = Flux
.just(“Lasagna”, “Lollipops”, “Apples”)
.delaySubscription(Duration.ofMillis(250))
.delayElements(Duration.ofMillis(500));
Flux mergedFlux = characterFlux.mergeWith(foodFlux);
mergedFlux.subscribe(s -> System.out.println(s));
StepVerifier.create(mergedFlux)
.expectNext(“Garfield”)
.expectNext(“Lasagna”)
.expectNext(“Kojak”)
.expectNext(“Lollipops”)
.expectNext(“Barbossa”)
.expectNext(“Apples”)
.verifyComplete();

Flux characterFlux = Flux
.just(“Garfield”, “Kojak”, “Barbossa”);
Flux foodFlux = Flux
.just(“Lasagna”, “Lollipops”, “Apples”);
Flux<Tuple2<String, String>> zippedFlux =
Flux.zip(characterFlux, foodFlux);
zippedFlux.subscribe(x -> System.out.println(x));
StepVerifier.create(zippedFlux)
.expectNextMatches(p ->
p.getT1().equals(“Garfield”) &&
p.getT2().equals(“Lasagna”))
.expectNextMatches(p ->
p.getT1().equals(“Kojak”) &&
p.getT2().equals(“Lollipops”))
.expectNextMatches(p ->
p.getT1().equals(“Barbossa”) &&
p.getT2().equals(“Apples”))
.verifyComplete();

Flux characterFlux = Flux
.just(“Garfield”, “Kojak”, “Barbossa”);
Flux foodFlux = Flux
.just(“Lasagna”, “Lollipops”, “Apples”);
Flux zippedFlux =
Flux.zip(characterFlux, foodFlux, (c, f) -> c + " eats " + f);
zippedFlux.subscribe(x -> System.out.println(x));
StepVerifier.create(zippedFlux)
.expectNext(“Garfield eats Lasagna”)
.expectNext(“Kojak eats Lollipops”)
.expectNext(“Barbossa eats Apples”)
.verifyComplete();
SELECTING THE FIRST REACTIVE TYPE TO PUBLISH

Flux slowFlux = Flux.just(“tortoise”, “snail”, “sloth”)
.delaySubscription(Duration.ofMillis(100));
Flux fastFlux = Flux.just(“hare”, “cheetah”, “squirrel”);
Flux firstFlux = Flux.first(slowFlux, fastFlux);
StepVerifier.create(firstFlux)
.expectNext(“hare”)
.expectNext(“cheetah”)
.expectNext(“squirrel”)
.verifyComplete();

FILTERING DATA FROM REACTIVE TYPES

Flux skipFlux = Flux.just(
“one”, “two”, “skip a few”, “ninety nine”, “one hundred”)
.skip(3);
StepVerifier.create(skipFlux)
.expectNext(“ninety nine”, “one hundred”)
.verifyComplete();

Flux skipFlux = Flux.just(
“one”, “two”, “skip a few”, “ninety nine”, “one hundred”)
.delayElements(Duration.ofSeconds(1))
.skip(Duration.ofSeconds(4));
StepVerifier.create(skipFlux)
.expectNext(“ninety nine”, “one hundred”)
.verifyComplete();

Flux nationalParkFlux = Flux.just(
“Yellowstone”, “Yosemite”, “Grand Canyon”,
“Zion”, “Grand Teton”)
.take(3);
StepVerifier.create(nationalParkFlux)
.expectNext(“Yellowstone”, “Yosemite”, “Grand Canyon”)
.verifyComplete();

Flux nationalParkFlux = Flux.just(
“Yellowstone”, “Yosemite”, “Grand Canyon”,
“Zion”, “Grand Teton”)
.delayElements(Duration.ofSeconds(1))
.take(Duration.ofMillis(3500));
StepVerifier.create(nationalParkFlux)
.expectNext(“Yellowstone”, “Yosemite”, “Grand Canyon”)
.verifyComplete();

Flux nationalParkFlux = Flux.just(
“Yellowstone”, “Yosemite”, “Grand Canyon”,
“Zion”, “Grand Teton”)
.filter(np -> !np.contains(" "));
StepVerifier.create(nationalParkFlux)
.expectNext(“Yellowstone”, “Yosemite”, “Zion”)
.verifyComplete();

Flux animalFlux = Flux.just(
“dog”, “cat”, “bird”, “dog”, “bird”, “anteater”)
.distinct();
StepVerifier.create(animalFlux)
.expectNext(“dog”, “cat”, “bird”, “anteater”)
.verifyComplete();
MAPPING REACTIVE DATA

Flux playerFlux = Flux
.just(“Michael Jordan”, “Scottie Pippen”, “Steve Kerr”)
.map(n -> {
String[] split = n.split("s");
return new Player(split[0], split[1]);
});
StepVerifier.create(playerFlux)
.expectNext(new Player(“Michael”, “Jordan”))
.expectNext(new Player(“Scottie”, “Pippen”))
.expectNext(new Player(“Steve”, “Kerr”))
.verifyComplete();

Flux playerFlux = Flux
.just(“Michael Jordan”, “Scottie Pippen”, “Steve Kerr”)
.flatMap(n -> Mono.just(n)
.map(p -> {
String[] split = p.split("s");
return new Player(split[0], split[1]);
})
.subscribeOn(Schedulers.parallel())
);
List playerList = Arrays.asList(
new Player(“Michael”, “Jordan”),
new Player(“Scottie”, “Pippen”),
new Player(“Steve”, “Kerr”));
StepVerifier.create(playerFlux)
.expectNextMatches(p -> playerList.contains§)
.expectNextMatches(p -> playerList.contains§)
.expectNextMatches(p -> playerList.contains§)
.verifyComplete();
BUFFERING DATA ON A REACTIVE STREAM

Flux fruitFlux = Flux.just(
“apple”, “orange”, “banana”, “kiwi”, “strawberry”);

Flux<List> bufferedFlux = fruitFlux.buffer(3);

StepVerifier
.create(bufferedFlux)
.expectNext(Arrays.asList(“apple”, “orange”, “banana”))
.expectNext(Arrays.asList(“kiwi”, “strawberry”))
.verifyComplete();

Buffering values from a reactive Flux into non-reactive List collections seems counterproductive. But when you combine buffer() with flatMap(), it enables each of the List collections to be processed in parallel:
Flux.just(
“apple”, “orange”, “banana”, “kiwi”, “strawberry”)
.buffer(3)
.flatMap(x ->
Flux.fromIterable(x)
.map(y -> y.toUpperCase())
.subscribeOn(Schedulers.parallel())
.log()
).subscribe();

Flux fruitFlux = Flux.just(
“apple”, “orange”, “banana”, “kiwi”, “strawberry”);

Mono<List> fruitListMono = fruitFlux.collectList();

StepVerifier
.create(fruitListMono)
.expectNext(Arrays.asList(
“apple”, “orange”, “banana”, “kiwi”, “strawberry”))
.verifyComplete();

Flux animalFlux = Flux.just(
“aardvark”, “elephant”, “koala”, “eagle”, “kangaroo”);

Mono<Map<Character, String>> animalMapMono =
animalFlux.collectMap(a -> a.charAt(0));

StepVerifier
.create(animalMapMono)
.expectNextMatches(map -> {
return
map.size() == 3 &&
map.get(‘a’).equals(“aardvark”) &&
map.get(‘e’).equals(“eagle”) &&
map.get(‘k’).equals(“kangaroo”);
})
.verifyComplete();

Performing logic operations on reactive types
Flux animalFlux = Flux.just(
“aardvark”, “elephant”, “koala”, “eagle”, “kangaroo”);

Mono hasAMono = animalFlux.all(a -> a.contains(“a”));
StepVerifier.create(hasAMono)
.expectNext(true)
.verifyComplete();

Mono hasKMono = animalFlux.all(a -> a.contains(“k”));
StepVerifier.create(hasKMono)
.expectNext(false)
.verifyComplete();

Flux animalFlux = Flux.just(
“aardvark”, “elephant”, “koala”, “eagle”, “kangaroo”);

Mono hasAMono = animalFlux.any(a -> a.contains(“a”));

StepVerifier.create(hasAMono)
.expectNext(true)
.verifyComplete();

Mono hasZMono = animalFlux.any(a -> a.contains(“z”));
StepVerifier.create(hasZMono)
.expectNext(false)
.verifyComplete();
Spring MVC change to Spring WebFlux

@GetMapping("/recent")
public Iterable recentTacos() {
PageRequest page = PageRequest.of(
0, 12, Sort.by(“createdAt”).descending());
return tacoRepo.findAll(page).getContent();
}

@GetMapping("/recent")
public Flux recentTacos() {
return Flux.fromIterable(tacoRepo.findAll()).take(12);
}

@PostMapping(consumes=“application/json”)
@ResponseStatus(HttpStatus.CREATED)
public Taco postTaco(@RequestBody Taco taco) {
return tacoRepo.save(taco);
}
@PostMapping(consumes=“application/json”)
@ResponseStatus(HttpStatus.CREATED)
public Mono postTaco(@RequestBody Mono tacoMono) {
return tacoRepo.saveAll(tacoMono).next();
}

public interface TacoRepository
extends ReactiveCrudRepository<Taco, Long> {
}
@GetMapping("/{id}")
public Taco tacoById(@PathVariable(“id”) Long id) {
Optional optTaco = tacoRepo.findById(id);
if (optTaco.isPresent()) {
return optTaco.get();
}
return null;
}
@GetMapping("/{id}")
public Mono tacoById(@PathVariable(“id”) Long id) {
return tacoRepo.findById(id);
}
WORKING WITH RXJAVA TYPES

@GetMapping("/recent")
public Observable recentTacos() {
return tacoService.getRecentTacos();
}

@GetMapping("/{id}")
public Single tacoById(@PathVariable(“id”) Long id) {
return tacoService.lookupTaco(id);
}
Developing Reactive APIs

@Configuration
public class RouterFunctionConfig {
@Autowired
private TacoRepository tacoRepo;
@Bean
public RouterFunction<?> routerFunction() {
return route(GET("/design/taco"), this::recents)
Testing reactive controllers 279
.andRoute(POST("/design"), this::postTaco);
}
public Mono recents(ServerRequest request) {
return ServerResponse.ok()
.body(tacoRepo.findAll().take(12), Taco.class);
}
public Mono postTaco(ServerRequest request) {
Mono taco = request.bodyToMono(Taco.class);
Mono savedTaco = tacoRepo.save(taco);
return ServerResponse
.created(URI.create(
“http://localhost:8080/design/taco/” +
savedTaco.getId()))
.body(savedTaco, Taco.class);
}
}
Test Reactive Rest APIs

// Test Get Method
Taco[] tacos = {
testTaco(1L), testTaco(2L),
testTaco(3L), testTaco(4L),
testTaco(5L), testTaco(6L),
testTaco(7L), testTaco(8L),
testTaco(9L), testTaco(10L),
testTaco(11L), testTaco(12L),
testTaco(13L), testTaco(14L),
testTaco(15L), testTaco(16L)};
Flux tacoFlux = Flux.just(tacos);
TacoRepository tacoRepo = Mockito.mock(TacoRepository.class);
when(tacoRepo.findAll()).thenReturn(tacoFlux);
WebTestClient testClient = WebTestClient.bindToController(
new DesignTacoController(tacoRepo))
.build();
testClient.get().uri("/design/recent")
.exchange()
.expectStatus().isOk()
.expectBody()
.jsonPath("" ) . i s A r r a y ( ) . j s o n P a t h ( " ").isArray() .jsonPath("").isArray().jsonPath("").isNotEmpty()
.jsonPath("[ 0 ] . i d " ) . i s E q u a l T o ( t a c o s [ 0 ] . g e t I d ( ) . t o S t r i n g ( ) ) . j s o n P a t h ( " [0].id").isEqualTo(tacos[0].getId().toString()) .jsonPath("[0].id").isEqualTo(tacos[0].getId().toString()).jsonPath("[0].name").isEqualTo(“Taco 1”).jsonPath("[ 1 ] . i d " ) . i s E q u a l T o ( t a c o s [ 1 ] . g e t I d ( ) . t o S t r i n g ( ) ) . j s o n P a t h ( " [1].id") .isEqualTo(tacos[1].getId().toString()).jsonPath("[1].id").isEqualTo(tacos[1].getId().toString()).jsonPath("[1].name")
.isEqualTo(“Taco 2”).jsonPath("[ 11 ] . i d " ) . i s E q u a l T o ( t a c o s [ 11 ] . g e t I d ( ) . t o S t r i n g ( ) ) . j s o n P a t h ( " [11].id") .isEqualTo(tacos[11].getId().toString()) .jsonPath("[11].id").isEqualTo(tacos[11].getId().toString()).jsonPath("[11].name").isEqualTo(“Taco 12”).jsonPath("[ 12 ] " ) . d o e s N o t E x i s t ( ) . j s o n P a t h ( " [12]") .doesNotExist().jsonPath("[12]").doesNotExist().jsonPath("[12]").doesNotExist();

// Test POST Method

TacoRepository tacoRepo = Mockito.mock(
TacoRepository.class);
Mono unsavedTacoMono = Mono.just(testTaco(null));
Taco savedTaco = testTaco(null);
savedTaco.setId(1L);
Mono savedTacoMono = Mono.just(savedTaco);
when(tacoRepo.save(any())).thenReturn(savedTacoMono);
WebTestClient testClient = WebTestClient.bindToController(
new DesignTacoController(tacoRepo)).build();
testClient.post()
.uri("/design")
.contentType(MediaType.APPLICATION_JSON)
.body(unsavedTacoMono, Taco.class)
.exchange()
.expectStatus().isCreated()
.expectBody(Taco.class)
.isEqualTo(savedTaco);

// Testing with a live server
@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment=WebEnvironment.RANDOM_PORT)
public class DesignTacoControllerWebTest {
@Autowired
private WebTestClient testClient;
@Test
public void shouldReturnRecentTacos() throws IOException {
testClient.get().uri("/design/recent")
.accept(MediaType.APPLICATION_JSON).exchange()
.expectStatus().isOk()
.expectBody()
.jsonPath("[ ? ( @ . i d = = ′ T A C O 1 ′ ) ] . n a m e " ) . i s E q u a l T o ( " C a r n i v o r e " ) . j s o n P a t h ( " [?(@.id == 'TACO1')].name") .isEqualTo("Carnivore") .jsonPath("[?(@.id==′TACO1′)].name").isEqualTo("Carnivore").jsonPath("[?(@.id == ‘TACO2’)].name")
.isEqualTo(“Bovine Bounty”)
.jsonPath("$[?(@.id == ‘TACO3’)].name")
.isEqualTo(“Veg-Out”);
}
}
Consume Reactive APIs

Flux ingredients = WebClient.create()
.get()
.uri(“http://localhost:8080/ingredients”)
.retrieve()
.bodyToFlux(Ingredient.class);
ingredients.subscribe(i -> { …})

Flux ingredients = WebClient.create()
.get()
.uri(“http://localhost:8080/ingredients”)
.retrieve()
.bodyToFlux(Ingredient.class);
ingredients
.timeout(Duration.ofSeconds(1))
.subscribe(
i -> { … },
e -> {
// handle timeout error
})

//Handing errors
ingredientMono.subscribe(
ingredient -> {
// handle the ingredient data
…
},
error-> {
// deal with the error
…
});

Mono ingredientMono = webClient
.get()
.uri(“http://localhost:8080/ingredients/{id}”, ingredientId)
.retrieve()
.onStatus(HttpStatus::is4xxClientError,
response -> Mono.just(new UnknownIngredientException()))
.bodyToMono(Ingredient.class);
Java 9
jshell

无法用单个下划线作为变量名称

int _ = 3; // java9 or above , error
String a = Objects.requireNonNullElse(m,“Bc”); // 若m不为null，则a = m，若m为null，则a = “Bc”
-cp, -classpath, --class-path(Java9新增)
Multi-Release JAR Files

–release
–class-path instead of -classpath
–version instead of -version
–module-path option has a shortcut -p
更多，见jeps

Java8中，接口可以有静态方法的默认实现，例：

public interface Test {
public static void print() {
System.out.println(“interface print”);
}

default void pout() {
    System.out.println();
}

• 1
• 2
• 3

}
Java9中，可以支持private的静态方法实现。

public interface Test {
private static void print() {
System.out.println(“interface print”);
}

static void pout() {
    print();
}

• 1
• 2
• 3

}
Optional.ofNullable(date).orElseGet(() -> newDate()); // date为null，才会执行newDate()方法，否则不执行newDate()方法
Optional.ofNullable(date).orElse(newDate()); // 无论date是否为null，都会执行newDate()方法
Java7中，可以使用try-with-Resources

try(Resouce res = …) {
work with res
}
res.close()会被自动执行

例：

try (var in = new Scanner(new FileInputStream(“C:UsersYoungDesktop新建文件夹1.tx.txt”), StandardCharsets.UTF_8);
var out = new PrintWriter(“C:UsersYoungDesktop新建文件夹out.txt”, StandardCharsets.UTF_8)) {
while (in.hasNext()) {
out.println(in.next().toUpperCase());
}
}
in 和 out执行完毕后都会自动关闭资源

在Java9 中，你可以在try中预先声明资源
例：

public static void printAll(String[] lines, PrintWriter out) {
try (out) { // effectively final variable
for (String line : lines) {
out.println(line);
} // out.close() called here
}
}
StackWalker用法示例

public class App {
/**
* Computes the factorial of a number
*
* @param n a non-negative integer
* @return n! = 1 * 2 * . . . * n
*/
public static int factorial(int n) {
System.out.println(“factorial(” + n + “):”);
var walker = StackWalker.getInstance();
walker.forEach(System.out::println);
int r;
if (n <= 1) {
r = 1;
} else {
r = n * factorial(n - 1);
}
System.out.println("return " + r);
return r;
}

public static void main(String[] args) {
    try (var in = new Scanner(System.in)) {
        System.out.print("Enter n: ");
        int n = in.nextInt();
        factorial(n);
    }
}

• 1
• 2
• 3
• 4
• 5
• 6
• 7

}
Java 9 expands the use of the diamond syntax to situations where it was previously not accepted. For example , you can now use diamonds with anonymous subclasses.

ArrayList list = new ArrayList<>(){
@Override
public String get(int index) {
return super.get(index).replaceAll(".","*");
}
};
Java 10
无需定义变量类型，通过var关键字+初始化的值，可以推测出变量类型

var a = 2; // a表示int
var b = “hello”; // b 表示String
var date = new java.util.Date();
var obj = new Custome(); // 自定义对象
Java 11
String repeated = “Java”.repeat(3); // 三个Java字符串连接