本篇文章給大家分享的是有關(guān)SpringBoot中RSocket如何使用，小編覺(jué)得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話(huà)不多說(shuō)，跟著小編一起來(lái)看看吧。

1. 概述

RSocket 應(yīng)用層協(xié)議支持Reactive Streams 語(yǔ)義， 例如：用RSocket作為HTTP的一種替代方案。在本教程中， 我們將看到RSocket 用在spring boot中，特別是spring boot 如何幫助抽象出更低級(jí)別的RSocket API。

2. 依賴(lài)

讓我們從添加spring-boot-starter-rsocket 依賴(lài)開(kāi)始：

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

這個(gè)依賴(lài)會(huì)傳遞性的拉取RSocket 相關(guān)的依賴(lài)，比如：rsocket-core 和rsocket-transport-netty

3.示例的應(yīng)用程序

現(xiàn)在繼續(xù)我們的簡(jiǎn)單應(yīng)用程序。為了突出RSocket 提供的交互模式，我打算創(chuàng)建一個(gè)交易應(yīng)用程序， 交易應(yīng)用程序包括客戶(hù)端和服務(wù)器。

3.1. 服務(wù)器設(shè)置

首先，我們?cè)O(shè)置由springboot應(yīng)用程序引導(dǎo)的RSocket server 服務(wù)器。 因?yàn)橛衧pring-boot-starter-rsocket dependency 依賴(lài)，所以springboot會(huì)自動(dòng)配置RSocket server 。 跟平常一樣， 可以用屬性驅(qū)動(dòng)的方式修改RSocket server 默認(rèn)配置值。例如：通過(guò)增加如下配置在application.properties 中，來(lái)修改RSocket 端口

spring.rsocket.server.port=7000

也可以根據(jù)需要進(jìn)一步修改服務(wù)器的其他屬性

3.2.設(shè)置客戶(hù)端

接下來(lái)，我們來(lái)設(shè)置客戶(hù)端，也是一個(gè)springboot應(yīng)用程序。雖然springboot自動(dòng)配置大部分RSocket相關(guān)的組件，但還要自定義一些對(duì)象來(lái)完成設(shè)置。

@Configurationpublic class ClientConfiguration { @Bean public RSocket rSocket() { return RSocketFactory .connect() .mimeType(MimeTypeUtils.APPLICATION_JSON_VALUE, MimeTypeUtils.APPLICATION_JSON_VALUE) .frameDecoder(PayloadDecoder.ZERO_COPY) .transport(TcpClientTransport.create(7000)) .start() .block(); } @Bean RSocketRequester rSocketRequester(RSocketStrategies rSocketStrategies) { return RSocketRequester.wrap(rSocket(), MimeTypeUtils.APPLICATION_JSON, rSocketStrategies); }}

這兒我們正在創(chuàng)建RSocket 客戶(hù)端并且配置TCP端口為：7000。注意： 該服務(wù)端口我們?cè)谇懊嬉呀?jīng)配置過(guò)。 接下來(lái)我們定義了一個(gè)RSocket的裝飾器對(duì)象RSocketRequester 。 這個(gè)對(duì)象在我們跟RSocket server 交互時(shí)會(huì)為我們提供幫助。 定義這些對(duì)象配置后，我們還只是有了一個(gè)骨架。在接下來(lái)，我們將暴露不同的交互模式， 并看看springboot在這個(gè)地方提供幫助的。

4.springboot RSocket 中的Request/Response

我們從Request/Response 開(kāi)始，HTTP 也使用這種通信方式，這也是最常見(jiàn)的、最相似的交互模式。 在這種交互模式里， 由客戶(hù)端初始化通信并發(fā)送一個(gè)請(qǐng)求。之后，服務(wù)器端執(zhí)行操作并返回一個(gè)響應(yīng)給客戶(hù)端--這時(shí)通信完成。 在我們的交易應(yīng)用程序里， 一個(gè)客戶(hù)端詢(xún)問(wèn)一個(gè)給定的股票的當(dāng)前的市場(chǎng)數(shù)據(jù)。 作為回復(fù)，服務(wù)器會(huì)傳遞請(qǐng)求的數(shù)據(jù)。

4.1.服務(wù)器

在服務(wù)器這邊，我們首先應(yīng)該創(chuàng)建一個(gè)controller 來(lái)持有我們的處理器方法。 我們會(huì)使用@MessageMapping 注解來(lái)代替像SpringMVC中的@RequestMapping 或者@GetMapping 注解

@Controllerpublic 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()); }}

來(lái)研究下我們的控制器。 我們將使用@Controller 注解來(lái)定義一個(gè)控制器來(lái)處理進(jìn)入RSocket的請(qǐng)求。 另外，注解@MessageMapping 讓我們定義我們感興趣的路由和如何響應(yīng)一個(gè)請(qǐng)求。 在這個(gè)示例中， 服務(wù)器監(jiān)聽(tīng)路由currentMarketData ， 并響應(yīng)一個(gè)單一的結(jié)果Mono<MarketData> 給客戶(hù)端。

4.2. 客戶(hù)端

接下來(lái)， 我們的RSocket客戶(hù)端應(yīng)該詢(xún)問(wèn)一只股票的價(jià)格并得到一個(gè)單一的響應(yīng)。 為了初始化請(qǐng)求， 我們?cè)撌褂肦SocketRequester 類(lèi)，如下：

@RestControllerpublic class MarketDataRestController { private final RSocketRequester rSocketRequester; public MarketDataRestController(RSocketRequester rSocketRequester) { this.rSocketRequester = rSocketRequester; } @GetMapping(value = "/current/{stock}") public Publisher<MarketData> current(@PathVariable("stock") String stock) { return rSocketRequester .route("currentMarketData") .data(new MarketDataRequest(stock)) .retrieveMono(MarketData.class); }}

注意：在示例中，RSocket 客戶(hù)端也是一個(gè)REST 風(fēng)格的controller ，以此來(lái)訪(fǎng)問(wèn)我們的RSocket 服務(wù)器。因此，我們使用@RestController 和@GetMapping 注解來(lái)定義我們的請(qǐng)求/響應(yīng)端點(diǎn)。 在端點(diǎn)方法中， 我們使用的是類(lèi)RSocketRequester 并指定了路由。 事實(shí)上，這個(gè)是服務(wù)器端RSocket 所期望的路由，然后我們傳遞請(qǐng)求數(shù)據(jù)。最后，當(dāng)調(diào)用retrieveMono() 方法時(shí)，springboot會(huì)幫我們初始化一個(gè)請(qǐng)求/響應(yīng)交互。

5.Spring Boot RSocket 中的Fire And Forget 模式

接下來(lái)我們將查看Fire And Forget 交互模式。正如名字提示的一樣，客戶(hù)端發(fā)送一個(gè)請(qǐng)求給服務(wù)器，但是不期望服務(wù)器的返回響應(yīng)回來(lái)。 在我們的交易程序中， 一些客戶(hù)端會(huì)作為數(shù)據(jù)資源服務(wù)，并且推送市場(chǎng)數(shù)據(jù)給服務(wù)器端。

5.1.服務(wù)器端

我們來(lái)創(chuàng)建另外一個(gè)端點(diǎn)在我們的服務(wù)器應(yīng)用程序中，如下：

@MessageMapping("collectMarketData")public Mono<Void> collectMarketData(MarketData marketData) { marketDataRepository.add(marketData); return Mono.empty();}

我們又一次定義了一個(gè)新的@MessageMapping 路由為collectMarketData 。此外， Spring Boot自動(dòng)轉(zhuǎn)換傳入的負(fù)載為一個(gè)MarketData 實(shí)例。 但是，這兒最大的不同是我們返回一個(gè)Mono<Void> ，因?yàn)榭蛻?hù)端不需要服務(wù)器的返回。

5.2. 客戶(hù)端

來(lái)看看我們?nèi)绾纬跏蓟覀兊膄ire-and-forget 模式的請(qǐng)求。 我們將創(chuàng)建另外一個(gè)REST風(fēng)格的端點(diǎn)，如下：

@GetMapping(value = "/collect")public Publisher<Void> collect() { return rSocketRequester .route("collectMarketData") .data(getMarketData()) .send();}

這兒我們指定路由和負(fù)載將是一個(gè)MarketData 實(shí)例。 由于我們使用send() 方法來(lái)代替retrieveMono() ，所有交互模式變成了fire-and-forget 模式。

6.Spring Boot RSocket 中的Request Stream

請(qǐng)求流是一種更復(fù)雜的交互模式， 這個(gè)模式中客戶(hù)端發(fā)送一個(gè)請(qǐng)求，但是在一段時(shí)間內(nèi)從服務(wù)器端獲取到多個(gè)響應(yīng)。 為了模擬這種交互模式， 客戶(hù)端會(huì)詢(xún)問(wèn)給定股票的所有市場(chǎng)數(shù)據(jù)。

6.1.服務(wù)器端

我們從服務(wù)器端開(kāi)始。 我們將添加另外一個(gè)消息映射方法，如下：

@MessageMapping("feedMarketData")public Flux<MarketData> feedMarketData(MarketDataRequest marketDataRequest) { return marketDataRepository.getAll(marketDataRequest.getStock());}

正如所見(jiàn)， 這個(gè)處理器方法跟其他的處理器方法非常類(lèi)似。 不同的部分是我們返回一個(gè)Flux<MarketData> 來(lái)代替Mono<MarketData> 。 最后我們的RSocket服務(wù)器會(huì)返回多個(gè)響應(yīng)給客戶(hù)端。

6.2.客戶(hù)端

在客戶(hù)端這邊， 我們?cè)搫?chuàng)建一個(gè)端點(diǎn)來(lái)初始化請(qǐng)求/響應(yīng)通信，如下：

@GetMapping(value = "/feed/{stock}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)public Publisher<MarketData> feed(@PathVariable("stock") String stock) { return rSocketRequester .route("feedMarketData") .data(new MarketDataRequest(stock)) .retrieveFlux(MarketData.class);}

我們來(lái)研究下RSocket請(qǐng)求。 首先我們定義了路由和請(qǐng)求負(fù)載。 然后，我們定義了使用retrieveFlux() 調(diào)用的響應(yīng)期望。這部分決定了交互模式。 另外注意：由于我們的客戶(hù)端也是REST 風(fēng)格的服務(wù)器，客戶(hù)端也定義了響應(yīng)媒介類(lèi)型MediaType.TEXT_EVENT_STREAM_VALUE 。

7.異常的處理

現(xiàn)在讓我們看看在服務(wù)器程序中，如何以聲明式的方式處理異常。 當(dāng)處理請(qǐng)求/響應(yīng)式， 我可以簡(jiǎn)單的使用@MessageExceptionHandler 注解，如下：

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

這里我們給異常處理方法標(biāo)記注解為@MessageExceptionHandler 。作為結(jié)果， 這個(gè)方法將處理所有類(lèi)型的異常， 因?yàn)镋xception 是所有其他類(lèi)型的異常的超類(lèi)。 我們也可以明確地創(chuàng)建更多的不同類(lèi)型的，不同的異常處理方法。 這當(dāng)然是請(qǐng)求/響應(yīng)模式，并且我們返回的是Mono<MarketData> 。我們期望這里的響應(yīng)類(lèi)型跟我們的交互模式的返回類(lèi)型相匹配。

總結(jié)

以上是生活随笔為你收集整理的SpringBoot中RSocket如何使用的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。