Springboot2.2中的RSocket体验

什么是RSocket?

RSocket是一种二进制协议，用于TCP、websocket和Aeron等字节流传输。它通过异步消息在单个连接上传递，支持以下对称交互模型:

request/response (stream of 1)
request/stream (finite stream of many)
fire-and-forget (no response)
channel (bi-directional streams)

一个发展过程

下面是一个简单的demo,包括两个模块producer和consumer供大家学习.

producer

核心pom.xml

       <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-rsocket</artifactId></dependency>

Repository

@Slf4j
@Component
public class MarketDataRepository {private static final int BOUND = 100;private Random random = new Random();public Flux<MarketData> getAll(String stock) {return Flux.fromStream(Stream.generate(() -> getMarketDataResponse(stock))).log().delayElements(Duration.ofSeconds(1));}public Mono<MarketData> getOne(String stock) {return Mono.just(getMarketDataResponse(stock));}public void add(MarketData marketData) {log.info("New market data: {}", marketData);}private MarketData getMarketDataResponse(String stock) {return new MarketData(stock, random.nextInt(BOUND));}
}

这个MarketDataRepository是一个简易模拟,当然你也可以使用R2DBC猛戳

controller

@Controller
public class MarketDataRSocketController {private final MarketDataRepository marketDataRepository;public MarketDataRSocketController(MarketDataRepository marketDataRepository) {this.marketDataRepository = marketDataRepository;}@MessageMapping("currentMarketData")public Mono<MarketData> currentMarketData(MarketDataRequest marketDataRequest) {return marketDataRepository.getOne(marketDataRequest.getStock());}@MessageMapping("feedMarketData")public Flux<MarketData> feedMarketData(MarketDataRequest marketDataRequest) {return marketDataRepository.getAll(marketDataRequest.getStock());}@MessageMapping("collectMarketData")public Mono<Void> collectMarketData(MarketData marketData) {marketDataRepository.add(marketData);return Mono.empty();}@MessageExceptionHandlerpublic Mono<MarketData> handleException(Exception e) {return Mono.just(MarketData.fromException(e));}
}

这个Controller用来处理消费者的RSocket请求,
注意这里的@MessageMapping其实可以理解为就是@RequestMapping,其实就是提供路由.

application.properties

spring.rsocket.server.port=7000

配置,目前只有四个,可以参考这里

异常处理

@MessageExceptionHandler
public Mono<MarketData> handleException(Exception e) {return Mono.just(MarketData.fromException(e));
}

注意:不同的模式需要不同的异常处理,这里是针对请求/响应的异常处理,啦啦啦.

consumer

核心pom.xml

       <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-rsocket</artifactId></dependency>

配置Configuration

@Lazy
@Configuration
public class ConsumerConfiguration {@BeanRSocket rSocket() {return RSocketFactory.connect().dataMimeType(MimeTypeUtils.APPLICATION_JSON_VALUE).frameDecoder(PayloadDecoder.ZERO_COPY)//.transport(TcpClientTransport.create("47.96.70.206",7000)) 个人服务器已关闭.transport(TcpClientTransport.create("127.0.0.1",7000)).start().block();}@BeanRSocketRequester requester(RSocketStrategies rSocketStrategies) {//废弃版//RSocketRequester.create(this.rSocket(),MimeTypeUtils.APPLICATION_JSON, rSocketStrategies);return  RSocketRequester.wrap(this.rSocket(),MimeTypeUtils.APPLICATION_JSON, rSocketStrategies);}
}

这里，我们创建RSocket Client(消费者)，并将其配置端口7000,使用TCP传输。注意，这是我们之前配置的服务器端口。
接下来，我们将定义一个RSocketRequester的实例，它是一个围绕RSocket的包装器。这个bean将帮助我们与RSocket服务器进行交互。
定义了这些bean配置之后，我们就有了一个基本的结构。

消费者Controller

@RestController
public class MarkDataRestController {private final RSocketRequester requester;public MarkDataRestController(RSocketRequester requester) {this.requester = requester;}/***功能描述 Request/Response模式,目前通用的请求/响应模式* @author KL* @date 2019/5/27* @param stock* @return org.reactivestreams.Publisher<com.github.consumer.entity.MarketData>*/@GetMapping("/current/{stock}")public Publisher<MarketData> current(@PathVariable String stock){return requester.route("currentMarketData").data(new MarketDataRequest(stock)).retrieveMono(MarketData.class);}/***功能描述 Request/Stream模式  ,一个简单的请求返回多个响应.* Request/Stream模式是一个更复杂的交互模型，其中客户机发送一个请求，但是在一段时间内从服务器获得多个响应。* @author KL* @date 2019/5/27* @param stock* @return org.reactivestreams.Publisher<com.github.consumer.entity.MarketData>*/@GetMapping(value = "/feed/{stock}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public Publisher<MarketData> feed(@PathVariable("stock") String stock) {return requester.route("feedMarketData").data(new MarketDataRequest(stock)).retrieveFlux(MarketData.class);}/***功能描述 Fire And Forget模式,其实就是Client推送给Server端* @author KL* @date 2019/5/27* @param* @return org.reactivestreams.Publisher<java.lang.Void>*/@GetMapping(value = "/collect")public Publisher<Void> collect() {return requester.route("collectMarketData").data(getMarketData()).send();}private MarketData getMarketData() {return new MarketData("X", new Random().nextInt(10));}
}

总结

RSocket为我们提供了一种新的服务通信思路,我们可以将RSocket应用到微服务框架中去,尝试去替代Http的方案.
现在Springboot支持了,在过不久dubbo3.0也支持了,哈哈.学习吧.