spring webflux返回application/stream+json
spring webflux返回application/stream+json
code4it
发布于 2018-09-17 15:53:42
发布于 2018-09-17 15:53:42
序
本文主要研究下spring webflux返回application/stream+json的实例
maven
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>controller
/**
* curl -i localhost:8080/stream
* @return
*/
@GetMapping(value = "/stream",produces = MediaType.APPLICATION_STREAM_JSON_VALUE)
public Flux<Price> priceStream(){ return Flux.interval(Duration.ofMillis(500)) .map(l -> new Price(System.currentTimeMillis(),ThreadLocalRandom.current().nextInt(100, 125))) .log();
}注意这里produces = MediaType.APPLICATION_STREAM_JSON_VALUE 如果不是application/stream+json则调用端无法滚动得到结果，将一直阻塞等待数据流结束或超时。
输出
- 后台输出
2018-02-08 21:36:49.701 INFO 1270 --- [ctor-http-nio-2] reactor.Flux.Map.1 : onSubscribe(FluxMap.MapSubscriber)2018-02-08 21:36:49.702 INFO 1270 --- [ctor-http-nio-2] reactor.Flux.Map.1 : request(1)2018-02-08 21:36:50.208 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097010208, value=120.0))2018-02-08 21:36:50.229 INFO 1270 --- [ctor-http-nio-2] reactor.Flux.Map.1 : request(31)2018-02-08 21:36:50.708 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097010708, value=124.0))2018-02-08 21:36:51.208 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097011208, value=119.0))2018-02-08 21:36:51.707 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097011707, value=120.0))2018-02-08 21:36:52.207 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097012207, value=109.0))2018-02-08 21:36:52.707 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097012707, value=101.0))2018-02-08 21:36:53.208 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097013208, value=114.0))2018-02-08 21:36:53.707 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097013707, value=113.0))2018-02-08 21:36:54.206 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097014206, value=105.0))2018-02-08 21:36:54.708 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097014708, value=103.0))2018-02-08 21:36:55.208 INFO 1270 --- [ parallel-2] reactor.Flux.Map.1 : onNext(Price(timestamp=1518097015207, value=123.0))2018-02-08 21:36:55.212 INFO 1270 --- [ctor-http-nio-2] reactor.Flux.Map.1 : cancel()- 调用端输出
curl -i localhost:8080/stream
HTTP/1.1 200 OK
transfer-encoding: chunked
Content-Type: application/stream+json;charset=UTF-8{"timestamp":1518097010208,"value":120.0}
{"timestamp":1518097010708,"value":124.0}
{"timestamp":1518097011208,"value":119.0}
{"timestamp":1518097011707,"value":120.0}
{"timestamp":1518097012207,"value":109.0}
{"timestamp":1518097012707,"value":101.0}
{"timestamp":1518097013208,"value":114.0}
{"timestamp":1518097013707,"value":113.0}
{"timestamp":1518097014206,"value":105.0}
^C可以看到由于使用了application/stream+json，返回的transfer-encoding是chunked，因此调用端可以做到滚动输出。
分页
使用了webflux之后，可能好奇之前的分页调用怎么办。reactive-streams是把数据当做数据流来用的，因此spring data reactive并不支持返回Page，但是调用参数可以传Pageable参数
public interface StocDao extends ReactiveCrudRepository<Stock, String> {
Flux<Stock> findByName(String name,Pageable pageable);
}注意这里返回Flux<Stock>，而不是Page<Stock> 也就是相当于丢失了total count
小结
对于webflux返回的Flux的流数据，需要配合返回MediaType.APPLICATION_STREAM_JSON_VALUE，同时调用端也需要能够支持这种mediaType(WebClient支持)，这样才能启到reactive streams的效果。
doc
- Serving large datasets with Spring WebFlux
- mongo.reactive.repositories.usage
- Is there any way to implement pagination in spring webflux and spring data reactive
本文参与 腾讯云自媒体分享计划，分享自微信公众号。
原始发表：2018-02-08，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录