1. 概述

目前創(chuàng)新互聯(lián)公司已為上千的企業(yè)提供了網(wǎng)站建設(shè)、域名、網(wǎng)絡(luò)空間、網(wǎng)站托管、服務(wù)器托管、企業(yè)網(wǎng)站設(shè)計、錫林郭勒盟網(wǎng)站維護等服務(wù)，公司將堅持客戶導(dǎo)向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長，共同發(fā)展。

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

2. 依賴

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

  org.springframework.boot
  spring-boot-starter-rsocket

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

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

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

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

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

spring.rsocket.server.port=7000

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

3.2.設(shè)置客戶端

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

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

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

4. springboot RSocket 中的 Request/Response

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

4.1.服務(wù)器

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

@Controller
public class MarketDataRSocketController {

  private final MarketDataRepository marketDataRepository;

  public MarketDataRSocketController(MarketDataRepository marketDataRepository) {
    this.marketDataRepository = marketDataRepository;
  }

  @MessageMapping("currentMarketData")
  public Mono currentMarketData(MarketDataRequest marketDataRequest) {
    return marketDataRepository.getOne(marketDataRequest.getStock());
  }
}

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

4.2. 客戶端

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

@RestController
public class MarketDataRestController {

  private final RSocketRequester rSocketRequester;

  public MarketDataRestController(RSocketRequester rSocketRequester) {
    this.rSocketRequester = rSocketRequester;
  }

  @GetMapping(value = "/current/{stock}")
  public Publisher current(@PathVariable("stock") String stock) {
    return rSocketRequester
     .route("currentMarketData")
     .data(new MarketDataRequest(stock))
     .retrieveMono(MarketData.class);
  }
}

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

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

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

5.1.服務(wù)器端

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

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

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

5.2. 客戶端

來看看我們?nèi)绾纬跏蓟覀兊?nbsp;fire-and-forget 模式的請求。 我們將創(chuàng)建另外一個REST風(fēng)格的端點，如下：

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

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

6. Spring Boot RSocket 中的 Request Stream

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

6.1.服務(wù)器端

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

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

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

6.2.客戶端

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

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

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

7.異常的處理

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

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

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

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持創(chuàng)新互聯(lián)。