@GetMapping("/reactive/restaurants")public Flux<Restaurant> findAll() {return restaurantRepository.findAll();}

使用webflux需要具备的基础是Reactive programming 的理解。
Reactor 的基础 和 熟练的java8 lambda使用

server.port=8081

@Data
public class Quote {private static final MathContext MATH_CONTEXT = new MathContext(2);private String ticker;private BigDecimal price;private Instant instant;public Quote() {}public Quote(String ticker, BigDecimal price) {this.ticker = ticker;this.price = price;}public Quote(String ticker, Double price) {this(ticker, new BigDecimal(price, MATH_CONTEXT));}@Overridepublic String toString() {return "Quote{" +"ticker='" + ticker + '\'' +", price=" + price +", instant=" + instant +'}';}
}

@Component
public class QuoteGenerator {private final MathContext mathContext = new MathContext(2);private final Random random = new Random();private final List<Quote> prices = new ArrayList<>();/*** 生成行情数据*/public QuoteGenerator() {this.prices.add(new Quote("CTXS", 82.26));this.prices.add(new Quote("DELL", 63.74));this.prices.add(new Quote("GOOG", 847.24));this.prices.add(new Quote("MSFT", 65.11));this.prices.add(new Quote("ORCL", 45.71));this.prices.add(new Quote("RHT", 84.29));this.prices.add(new Quote("VMW", 92.21));}public Flux<Quote> fetchQuoteStream(Duration period) {// 需要周期生成值并返回，使用 Flux.intervalreturn Flux.interval(period)// In case of back-pressure, drop events
                .onBackpressureDrop()// For each tick, generate a list of quotes.map(this::generateQuotes)// "flatten" that List<Quote> into a Flux<Quote>.flatMapIterable(quotes -> quotes).log("io.spring.workshop.stockquotes");}/*** Create quotes for all tickers at a single instant.*/private List<Quote> generateQuotes(long interval) {final Instant instant = Instant.now();return prices.stream().map(baseQuote -> {BigDecimal priceChange = baseQuote.getPrice().multiply(new BigDecimal(0.05 * this.random.nextDouble()), this.mathContext);Quote result = new Quote(baseQuote.getTicker(), baseQuote.getPrice().add(priceChange));result.setInstant(instant);return result;}).collect(Collectors.toList());}
}

@Component
public class QuoteHandler {private final Flux<Quote> quoteStream;public QuoteHandler(QuoteGenerator quoteGenerator) {this.quoteStream = quoteGenerator.fetchQuoteStream(ofMillis(1000)).share();}public Mono<ServerResponse> hello(ServerRequest request) {return ok().contentType(TEXT_PLAIN).body(BodyInserters.fromObject("Hello Spring!"));}public Mono<ServerResponse> echo(ServerRequest request) {return ok().contentType(TEXT_PLAIN).body(request.bodyToMono(String.class), String.class);}public Mono<ServerResponse> streamQuotes(ServerRequest request) {return ok().contentType(APPLICATION_STREAM_JSON).body(this.quoteStream, Quote.class);}public Mono<ServerResponse> fetchQuotes(ServerRequest request) {int size = Integer.parseInt(request.queryParam("size").orElse("10"));return ok().contentType(APPLICATION_JSON).body(this.quoteStream.take(size), Quote.class);}
}

@Configuration
public class QuoteRouter {@Beanpublic RouterFunction<ServerResponse> route(QuoteHandler quoteHandler) {return RouterFunctions.route(GET("/hello").and(accept(TEXT_PLAIN)), quoteHandler::hello).andRoute(POST("/echo").and(accept(TEXT_PLAIN).and(contentType(TEXT_PLAIN))), quoteHandler::echo).andRoute(GET("/quotes").and(accept(APPLICATION_JSON)), quoteHandler::fetchQuotes).andRoute(GET("/quotes").and(accept(APPLICATION_STREAM_JSON)), quoteHandler::streamQuotes);}
}

$ curl http://localhost:8081/echo -i -d "WebFlux workshop" -H "Content-Type: text/plain"
HTTP/1.1 200 OK
transfer-encoding: chunked
Content-Type: text/plainWebFlux workshop

$ curl http://localhost:8081/quotes -i -H "Accept: application/stream+json"
HTTP/1.1 200 OK
transfer-encoding: chunked
Content-Type: application/stream+json{"ticker":"CTXS","price":82.77,"instant":"2018-05-15T06:45:51.261Z"}
{"ticker":"DELL","price":64.83,"instant":"2018-05-15T06:45:51.261Z"}
{"ticker":"GOOG","price":881,"instant":"2018-05-15T06:45:51.261Z"}
{"ticker":"MSFT","price":67.3,"instant":"2018-05-15T06:45:51.261Z"}
{"ticker":"ORCL","price":48.1,"instant":"2018-05-15T06:45:51.261Z"}
{"ticker":"RHT","price":85.1,"instant":"2018-05-15T06:45:51.261Z"}
{"ticker":"VMW","price":92.24,"instant":"2018-05-15T06:45:51.261Z"}
-------------------------------无敌分割线-------------------------------------
{"ticker":"CTXS","price":85.7,"instant":"2018-05-15T06:45:52.260Z"}
{"ticker":"DELL","price":64.12,"instant":"2018-05-15T06:45:52.260Z"}
{"ticker":"GOOG","price":879,"instant":"2018-05-15T06:45:52.260Z"}
{"ticker":"MSFT","price":67.9,"instant":"2018-05-15T06:45:52.260Z"}
{"ticker":"ORCL","price":46.43,"instant":"2018-05-15T06:45:52.260Z"}
{"ticker":"RHT","price":86.8,"instant":"2018-05-15T06:45:52.260Z"}
...

@RunWith(SpringRunner.class)
//  We create a `@SpringBootTest`, starting an actual server on a `RANDOM_PORT`
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class StockQuotesApplicationTests {// Spring Boot will create a `WebTestClient` for you,// already configure and ready to issue requests against "localhost:RANDOM_PORT"
   @Autowiredprivate WebTestClient webTestClient;@Testpublic void fetchQuotes() {webTestClient// We then create a GET request to test an endpoint.get().uri("/quotes?size=20").accept(MediaType.APPLICATION_JSON).exchange()// and use the dedicated DSL to test assertions against the response
            .expectStatus().isOk().expectHeader().contentType(MediaType.APPLICATION_JSON).expectBodyList(Quote.class).hasSize(20)// Here we check that all Quotes have a positive price value.consumeWith(allQuotes ->assertThat(allQuotes.getResponseBody()).allSatisfy(quote -> assertThat(quote.getPrice()).isPositive()));}@Testpublic void fetchQuotesAsStream() {List<Quote> result = webTestClient// We then create a GET request to test an endpoint.get().uri("/quotes")// this time, accepting "application/stream+json"
            .accept(MediaType.APPLICATION_STREAM_JSON).exchange()// and use the dedicated DSL to test assertions against the response
            .expectStatus().isOk().expectHeader().contentType(MediaType.APPLICATION_STREAM_JSON).returnResult(Quote.class).getResponseBody().take(30).collectList().block();assertThat(result).allSatisfy(quote -> assertThat(quote.getPrice()).isPositive());}
}