原創：扣釘日記（微信公眾號ID：codelogs），歡迎分享，非公眾號轉載保留此聲明。在之前的OOM問題復盤中，我們添加了jmap腳本來自動dump記憶體現場，方便排查OOM問題。但當我反覆模擬OOM場景測試時，發現jmap有時可以dump成功，有時會報錯，如下：經過網上一頓搜索，發現兩種原因可 ...

響應式

響應式系統（Reactive System）

具有以下特質：即時響應性（Responsive）、回彈性（Resilient）、彈性（Elastic）以及消息驅動（Message Driven）響應式系統更加靈活，松耦合，可伸縮

即時響應性
只要有可能，系統就會及時地做出響應。即時響應是可用性和實用性的基石，而更加重要的是，即時響應意味著可以快速地檢測到問題並且有效地對其進行處理。即時響應的系統專註於提供快速而一致的響應時間，確立可靠的反饋上限，以提供一致的服務質量。這種一致的行為轉而將簡化錯誤處理、建立最終用戶的信任並促使用戶與系統作進一步的互動
回彈性
系統在出現失敗時依然保持即時響應性。這不僅適用於高可用的、任務關鍵型系統——任何不具備回彈性的系統都將會在發生失敗之後丟失即時響應性。回彈性是通過複製、遏制、隔離以及委托來實現的。失敗的擴散被遏制在了每個組件內部，與其他組件相互隔離，從而確保系統某部分的失敗不會危及整個系統，並能獨立恢復。每個組件的恢復都被委托給了另一個（外部的）組件，此外，在必要時可以通過複製來保證高可用性。（因此）組件的客戶端不再承擔組件失敗的處理
彈性
系統在不斷變化的工作負載之下依然保持即時響應性。響應式系統可以對輸入（負載）的速率變化做出反應，比如通過增加或者減少被分配用於服務這些輸入（負載）的資源。這意味著設計上並沒有爭用點和中央瓶頸，得以進行組件的分片或者複製，併在它們之間分佈輸入（負載）。通過提供相關的實時性能指標，響應式系統能支持預測式以及反應式的伸縮演算法。這些系統可以在常規的硬體以及軟體平臺上實現成本高效的彈性。
消息驅動
消息驅動：響應式系統依賴非同步的消息傳遞，從而確保了松耦合、隔離、位置透明的組件之間有著明確邊界。這一邊界還提供了將失敗作為消息委托出去的手段。使用顯式的消息傳遞，可以通過在系統中塑造並監視消息流隊列，併在必要時應用回壓，從而實現負載管理、彈性以及流量控制。使用位置透明的消息傳遞作為通信的手段，使得跨集群或者在單個主機中使用相同的結構成分和語義來管理失敗成為了可能。非阻塞的通信使得接收者可以只在活動時才消耗資源，從而減少系統開銷。

響應式編程

響應式編程（reactive programming）是一種基於數據流（data stream）和變化傳遞（propagation of change）的聲明式（declarative）的編程範式

變化傳遞
在命令式編程下，式子 a=b+c，代表a的值是由b和c計算出來的。如果b或c後續有變化，不會影響到a的值。
在響應式編程下，式子 a:=b+c，代表著a的值是由b和c計算出來的。但如果b或者c的值後續有變化，會影響到a的值。

響應式流

Reactive Streams致力於提供一套非阻塞背壓的非同步流處理標準規範
通過定義一組實體，介面和互操作方法，給出了實現非同步非阻塞背壓的標準。第三方遵循這個標準來實現具體的解決方案，常見的有Reactor，RxJava，Akka Streams，Ratpack等

響應式流定義 org.reactivestreams

Reactor

Reactor 是一個用於JVM的完全非阻塞的響應式編程框架，具備高效的需求管理（即對 “背壓（backpressure）”的控制）能力。它與 Java 8 函數式 API 直接集成，比如 CompletableFuture， Stream，以及 Duration。它提供了非同步序列 API Flux（用於[N]個元素）和 Mono（用於 [0|1]個元素），並完全遵循和實現了“響應式擴展規範”（Reactive Extensions Specification）

Flux

Flux 是一個能夠發出 0 到 N 個元素的標準的 Publisher，它會被一個“錯誤（error）” 或“完成（completion）”信號終止。因此，一個 flux 的可能結果是一個 value、completion 或 error。就像在響應式流規範中規定的那樣，這三種類型的信號被翻譯為面向下游的 onNext，onComplete和onError方法。

Mono

Mono 是一種特殊的 Publisher，它最多發出一個元素，然後終止於一個 onComplete 信號或一個 onError 信號。

Scheduler

在 Reactor 中，執行模式以及執行過程取決於所使用的 Scheduler。 Scheduler 是一個擁有廣泛實現類的抽象介面。Schedulers 類提供的靜態方法:

當前線程（Schedulers.immediate()）
可重用的單線程（Schedulers.single()）。註意，這個方法對所有調用者都提供同一個線程來使用，直到該調度器（Scheduler）被廢棄。如果你想使用專一的線程，就對每一個調用使用 Schedulers.newSingle()。
彈性線程池（Schedulers.elastic()。它根據需要創建一個線程池，重用空閑線程。線程池如果空閑時間過長（預設為 60s）就會被廢棄。對於 I/O 阻塞的場景比較適用。 Schedulers.elastic() 能夠方便地給一個阻塞的任務分配它自己的線程，從而不會妨礙其他任務和資源。
固定大小線程池（Schedulers.parallel()）。所創建線程池的大小與 CPU 個數等同。
自定義調度器，Schedulers.fromExecutorService(ExecutorService) 基於現有的 ExecutorService 創建 Scheduler。

發佈訂閱

publishOn 的用法和處於訂閱鏈（subscriber chain）中的其他操作符一樣。它將上游信號傳給下游，同時執行指定的調度器 Scheduler 的某個工作線程上的回調。它會改變後續的操作符的執行所線上程（直到下一個 publishOn 出現在這個鏈上）。
subscribeOn 用於訂閱（subscription）過程，作用於那個向上的訂閱鏈（發佈者在被訂閱時才激活，訂閱的傳遞方向是向上游的）。所以，無論你把 subscribeOn 至於操作鏈的什麼位置，它都會影響到源頭的線程執行環境（context）。但是，它不會影響到後續的 publishOn，後者仍能夠切換其後操作符的線程執行環境。
在訂閱（subscribe）前，只是定義了處理流程，而沒有啟動發佈者。基於此，Reactor 可以使用這些規則來決定如何執行操作鏈。然後，一旦被訂閱了，整個流程就開始工作了。

異常

onErrorReturn 返回靜態預設值
onErrorResume 捕獲並執行一個異常處理方法，動態計算一個候補值，將異常包裝後再次拋出
onErrorMap 捕獲，包裝一個新異常後再次拋出
doOnError 捕獲異常，並繼續拋出
retry 重試

Spring WebFlux

SpringWebFlux開發

// 處理請求參數
Mono<String> string = request.body(BodyExtractors.toMono(String.class));
Flux<Person> people = request.body(BodyExtractors.toFlux(Person.class));
Mono<MultiValueMap<String, String>> map = request.formData(); 
// 封裝響應參數
Mono<Person> person = ...
ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(person, Person.class);


public class PersonHandler {
    private final PersonRepository repository;
    public PersonHandler(PersonRepository repository) {
        this.repository = repository;
    }
    public Mono<ServerResponse> listPeople(ServerRequest request) { 
        Flux<Person> people = repository.allPeople();
        return ok().contentType(APPLICATION_JSON).body(people, Person.class);
    }
    public Mono<ServerResponse> createPerson(ServerRequest request) { 
        Mono<Person> person = request.bodyToMono(Person.class);
        return ok().build(repository.savePerson(person));
    }
    public Mono<ServerResponse> getPerson(ServerRequest request) { 
        int personId = Integer.valueOf(request.pathVariable("id"));
        return repository.getPerson(personId)
            .flatMap(person -> ok().contentType(APPLICATION_JSON).bodyValue(person))
            .switchIfEmpty(ServerResponse.notFound().build());
    }
}

PersonRepository repository = ...
PersonHandler handler = new PersonHandler(repository);

RouterFunction<ServerResponse> otherRoute = ...
//介面路由
RouterFunction<ServerResponse> route = route()
    .GET("/person/{id}", accept(APPLICATION_JSON), handler::getPerson) 
    .GET("/person", accept(APPLICATION_JSON), handler::listPeople) 
    .POST("/person", handler::createPerson) 
    .add(otherRoute) 
    .build();

Spring WebFlux 也可以完全復用SpringMVC的註解

@RestController
@RequestMapping("/users")
public class MyRestController {
    private final UserRepository userRepository;
    private final CustomerRepository customerRepository;
    public MyRestController(UserRepository userRepository, CustomerRepository customerRepository) {
        this.userRepository = userRepository;
        this.customerRepository = customerRepository;
    }
    @GetMapping("/{userId}")
    public Mono<User> getUser(@PathVariable Long userId) { return this.userRepository.findById(userId);}

    @GetMapping("/{userId}/customers")
    public Flux<Customer> getUserCustomers(@PathVariable Long userId) {
	return this.userRepository.findById(userId).flatMapMany(this.customerRepository::findByUser);
    }

    @DeleteMapping("/{userId}")
    public Mono<Void> deleteUser(@PathVariable Long userId) { return this.userRepository.deleteById(userId);}
}

SpringWebFlux 使用feign進行微服務調用

<dependency>
            <groupId>com.playtika.reactivefeign</groupId>
            <artifactId>feign-reactor-spring-cloud-starter</artifactId>
            <version>3.1.5</version>
            <type>pom</type>
</dependency>

@EnableReactiveFeignClients


@ReactiveFeignClient(value = "platform-domain-gateway-server")
public interface DomainGateWayClient {

    @PostMapping("/domain/service")
    Mono<ResponseEntity<Object>> createDomainService(JsonLoadParam param);

}

參考文章
https://www.reactiveprinciples.org/
https://www.reactivemanifesto.org/zh-CN
http://www.reactive-streams.org/
https://www.tutorialspoint.com/concurrency_in_python/concurrency_in_python_reactive_programming.htm
https://link.zhihu.com/?target=https%3A//gist.github.com/staltz/868e7e9bc2a7b8c1f754
https://projectreactor.io/
https://docs.spring.io/spring-framework/docs/current/reference/html/web-reactive.html

響應式編程初探

響應式