Spring Boot中使用RSocket

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

2. 依賴

讓我們從添加spring-boot-starter-rsocket依賴開始：

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

這個依賴會傳遞性的拉取RSocket相關的依賴，比如：rsocket-core 和 rsocket-transport-netty

3.示例的應用程序

現在繼續我們的簡單應用程序。為了突出RSocket提供的交互模式，我打算創建一個交易應用程序，交易應用程序包括客戶端和服務器。

3.1. 服務器設置

首先，我們設置由springboot應用程序引導的RSocket server服務器。因為有spring-boot-starter-rsocket dependency依賴，所以springboot會自動配置RSocket server。跟平常一樣，可以用屬性驅動的方式修改RSocket server默認配置值。例如：通過增加如下配置在application.properties中，來修改RSocket端口

spring.rsocket.server.port=7000

也可以根據需要進一步修改服務器的其他屬性

3.2.設置客戶端

接下來，我們來設置客戶端，也是一個springboot應用程序。雖然springboot自動配置大部分RSocket相關的組件，但還要自定義一些對象來完成設置。

@Configuration
public 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); } }

這兒我們正在創建RSocket客戶端並且配置TCP端口為：7000。注意：該服務端口我們在前面已經配置過。接下來我們定義了一個RSocket的裝飾器對象RSocketRequester。這個對象在我們跟RSocket server交互時會為我們提供幫助。定義這些對象配置后，我們還只是有了一個骨架。在接下來，我們將暴露不同的交互模式，並看看springboot在這個地方提供幫助的。

4. `springboot RSocket` 中的 `Request/Response`

我們從Request/Response開始，HTTP也使用這種通信方式，這也是最常見的、最相似的交互模式。在這種交互模式里，由客戶端初始化通信並發送一個請求。之后，服務器端執行操作並返回一個響應給客戶端--這時通信完成。在我們的交易應用程序里，一個客戶端詢問一個給定的股票的當前的市場數據。作為回復，服務器會傳遞請求的數據。

4.1.服務器

在服務器這邊，我們首先應該創建一個controller 來持有我們的處理器方法。我們會使用 @MessageMapping注解來代替像SpringMVC中的@RequestMapping或者@GetMapping注解

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

來研究下我們的控制器。我們將使用@Controller注解來定義一個控制器來處理進入RSocket的請求。另外，注解@MessageMapping讓我們定義我們感興趣的路由和如何響應一個請求。在這個示例中，服務器監聽路由currentMarketData，並響應一個單一的結果Mono<MarketData>給客戶端。

4.2. 客戶端

接下來，我們的RSocket客戶端應該詢問一只股票的價格並得到一個單一的響應。為了初始化請求，我們該使用RSocketRequester類，如下：

@RestController
public 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客戶端也是一個REST風格的controller，以此來訪問我們的RSocket服務器。因此，我們使用@RestController和@GetMapping注解來定義我們的請求/響應端點。在端點方法中，我們使用的是類RSocketRequester並指定了路由。事實上，這個是服務器端RSocket所期望的路由，然后我們傳遞請求數據。最后，當調用retrieveMono()方法時，springboot會幫我們初始化一個請求/響應交互。

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

接下來我們將查看 Fire And Forget交互模式。正如名字提示的一樣，客戶端發送一個請求給服務器，但是不期望服務器的返回響應回來。在我們的交易程序中，一些客戶端會作為數據資源服務，並且推送市場數據給服務器端。

5.1.服務器端

我們來創建另外一個端點在我們的服務器應用程序中，如下：

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

我們又一次定義了一個新的@MessageMapping路由為collectMarketData。此外， Spring Boot自動轉換傳入的負載為一個MarketData實例。但是，這兒最大的不同是我們返回一個Mono<Void>，因為客戶端不需要服務器的返回。

5.2. 客戶端

來看看我們如何初始化我們的fire-and-forget模式的請求。我們將創建另外一個REST風格的端點，如下：

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

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

6.`Spring Boot RSocket`中的`Request Stream`

請求流是一種更復雜的交互模式，這個模式中客戶端發送一個請求，但是在一段時間內從服務器端獲取到多個響應。為了模擬這種交互模式，客戶端會詢問給定股票的所有市場數據。

6.1.服務器端

我們從服務器端開始。我們將添加另外一個消息映射方法，如下：

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

正如所見，這個處理器方法跟其他的處理器方法非常類似。不同的部分是我們返回一個Flux<MarketData>來代替Mono<MarketData>。最后我們的RSocket服務器會返回多個響應給客戶端。

6.2.客戶端

在客戶端這邊，我們該創建一個端點來初始化請求/響應通信，如下：

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

我們來研究下RSocket請求。首先我們定義了路由和請求負載。然后，我們定義了使用retrieveFlux()調用的響應期望。這部分決定了交互模式。另外注意：由於我們的客戶端也是REST風格的服務器，客戶端也定義了響應媒介類型MediaType.TEXT_EVENT_STREAM_VALUE。

7.異常的處理

現在讓我們看看在服務器程序中，如何以聲明式的方式處理異常。當處理請求/響應式，我可以簡單的使用@MessageExceptionHandler注解，如下：

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

這里我們給異常處理方法標記注解為@MessageExceptionHandler。作為結果，這個方法將處理所有類型的異常，因為Exception是所有其他類型的異常的超類。我們也可以明確地創建更多的不同類型的，不同的異常處理方法。這當然是請求/響應模式，並且我們返回的是Mono<MarketData>。我們期望這里的響應類型跟我們的交互模式的返回類型相匹配。

8.總結

在本教程中，我們介紹了springboot的RSocket支持，並詳細列出了RSocket提供的不同交互模式。查看所有示例代碼在GitHub上。

原文鏈接：www.baeldung.com/spring-boot…

作者：baeldung

譯者：sleeve