怎么理解Java的回調(diào)與反應(yīng)模式

這篇文章主要介紹“怎么理解Java的回調(diào)與反應(yīng)模式”，在日常操作中，相信很多人在怎么理解Java的回調(diào)與反應(yīng)模式問題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”怎么理解Java的回調(diào)與反應(yīng)模式”的疑惑有所幫助！接下來，請(qǐng)跟著小編一起來學(xué)習(xí)吧！

1. 目標(biāo)服務(wù)

客戶端調(diào)用表示城市詳細(xì)信息的目標(biāo)服務(wù)有兩個(gè)端口。當(dāng)使用類型為——/cityids 的 URI 調(diào)用時(shí)，返回城市 id 列表，并且示例結(jié)果如下所示：

[    1,    2,    3,    4,    5,    6,    7]

一個(gè)端口返回給定其 ID 的城市的詳細(xì)信息，例如，當(dāng)使用 ID 為1——“/cities/1” 調(diào)用時(shí)：

{    "country": "USA",    "id": 1,    "name": "Portland",    "pop": 1600000}

客戶端的責(zé)任是獲取城市 ID 的列表，然后對(duì)于每個(gè)城市，根據(jù) ID 獲取城市的詳細(xì)信息并將其組合到城市列表中。

2. 同步調(diào)用

我正在使用 Spring Framework 的 RestTemplate 進(jìn)行遠(yuǎn)程調(diào)用。獲取 cityId 列表的 Kotlin 函數(shù)如下所示：

private fun getCityIds(): List<String> {    val cityIdsEntity: ResponseEntity<List<String>> = restTemplate            .exchange("http://localhost:$localServerPort/cityids",                    HttpMethod.GET,                    null,                    object : ParameterizedTypeReference<List<String>>() {})    return cityIdsEntity.body!!}

獲取城市詳情：

private fun getCityForId(id: String): City {    return restTemplate.getForObject("http://localhost:$localServerPort/cities/$id", City::class.java)!!}

鑒于這兩個(gè)函數(shù)，它們很容易組合，以便于輕松返回城市列表 ：

val cityIds: List<String> = getCityIds()val cities: List<City> = cityIds        .stream()        .map<City> { cityId -> getCityForId(cityId) }        .collect(Collectors.toList())cities.forEach { city -> LOGGER.info(city.toString()) }

代碼很容易理解；但是，涉及八個(gè)阻塞調(diào)用：

1. 獲取 7 個(gè)城市 ID 的列表，然后獲取每個(gè)城市的詳細(xì)信息

2. 獲取 7 個(gè)城市的詳細(xì)信息

每一個(gè)調(diào)用都將在不同的線程上。

3. 非阻塞 IO 回調(diào)

我將使用 AsyncHttpClient 庫(kù)來進(jìn)行非阻塞 IO 調(diào)用。

進(jìn)行遠(yuǎn)程調(diào)用時(shí)，AyncHttpClient 返回 ListenableFuture 類型。

val responseListenableFuture: ListenableFuture<Response> = asyncHttpClient                .prepareGet("http://localhost:$localServerPort/cityids")                .execute()

可以將回調(diào)附加到 ListenableFuture 以在可用時(shí)對(duì)響應(yīng)進(jìn)行操作。

responseListenableFuture.addListener(Runnable {    val response: Response = responseListenableFuture.get()    val responseBody: String = response.responseBody    val cityIds: List<Long> = objectMapper.readValue<List<Long>>(responseBody,            object : TypeReference<List<Long>>() {})    ....}

鑒于 cityIds 的列表，我想獲得城市的詳細(xì)信息，因此從響應(yīng)中，我需要進(jìn)行更多的遠(yuǎn)程調(diào)用并為每個(gè)調(diào)用附加回調(diào)以獲取城市的詳細(xì)信息：

val responseListenableFuture: ListenableFuture<Response> = asyncHttpClient        .prepareGet("http://localhost:$localServerPort/cityids")        .execute()responseListenableFuture.addListener(Runnable {    val response: Response = responseListenableFuture.get()    val responseBody: String = response.responseBody    val cityIds: List<Long> = objectMapper.readValue<List<Long>>(responseBody,            object : TypeReference<List<Long>>() {})    cityIds.stream().map { cityId ->        val cityListenableFuture = asyncHttpClient                .prepareGet("http://localhost:$localServerPort/cities/$cityId")                .execute()        cityListenableFuture.addListener(Runnable {            val cityDescResp = cityListenableFuture.get()            val cityDesc = cityDescResp.responseBody            val city = objectMapper.readValue(cityDesc, City::class.java)            LOGGER.info("Got city: $city")        }, executor)    }.collect(Collectors.toList())}, executor)

這是一段粗糙的代碼；回調(diào)中又包含一組回調(diào)，很難推理和理解 - 因此它被稱為“回調(diào)地獄”。

4. 在 Java CompletableFuture 中使用非阻塞 IO

通過將 Java 的 CompletableFuture 作為返回類型而不是 ListenableFuture 返回，可以稍微改進(jìn)此代碼。CompletableFuture 提供允許修改和返回類型的運(yùn)算符。

例如，考慮獲取城市 ID 列表的功能：

private fun getCityIds(): CompletableFuture<List<Long>> {    return asyncHttpClient            .prepareGet("http://localhost:$localServerPort/cityids")            .execute()            .toCompletableFuture()            .thenApply { response ->                val s = response.responseBody                val l: List<Long> = objectMapper.readValue(s, object : TypeReference<List<Long>>() {})                l            }}

在這里，我使用 thenApply 運(yùn)算符將 CompletableFuture<Response> 轉(zhuǎn)換為 CompletableFuture<List<Long>> 。

同樣的，獲取城市詳情：

private fun getCityDetail(cityId: Long): CompletableFuture<City> {    return asyncHttpClient.prepareGet("http://localhost:$localServerPort/cities/$cityId")            .execute()            .toCompletableFuture()            .thenApply { response ->                val s = response.responseBody                LOGGER.info("Got {}", s)                val city = objectMaper.readValue(s, City::class.java)                city            }}

這是基于回調(diào)的方法的改進(jìn)。但是，在這個(gè)特定情況下，CompletableFuture 缺乏有用的運(yùn)算符，例如，所有城市細(xì)節(jié)都需要放在一起：

val cityIdsFuture: CompletableFuture<List<Long>> = getCityIds()val citiesCompletableFuture: CompletableFuture<List<City>> =        cityIdsFuture                .thenCompose { l ->                    val citiesCompletable: List<CompletableFuture<City>> =                            l.stream()                                    .map { cityId ->                                        getCityDetail(cityId)                                    }.collect(toList())                    val citiesCompletableFutureOfList: CompletableFuture<List<City>> = CompletableFuture.allOf(*citiesCompletable.toTypedArray())                                    .thenApply { _: Void? ->                                        citiesCompletable                                                .stream()                                                .map { it.join() }                                                .collect(toList())                                    }                    citiesCompletableFutureOfList                }

使用了一個(gè)名為 CompletableFuture.allOf 的運(yùn)算符，它返回一個(gè)“Void”類型，并且必須強(qiáng)制返回所需類型的 CompletableFuture<List<City>>。

5. 使用 Reactor 項(xiàng)目

Project Reactor 是 Reactive Streams 規(guī)范的實(shí)現(xiàn)。它有兩種特殊類型可以返回 0/1 項(xiàng)的流和 0/n 項(xiàng)的流 - 前者是 Mono，后者是 Flux。

Project Reactor 提供了一組非常豐富的運(yùn)算符，允許以各種方式轉(zhuǎn)換數(shù)據(jù)流。首先考慮返回城市 ID 列表的函數(shù)：

private fun getCityIds(): Flux<Long> {    return webClient.get()            .uri("/cityids")            .exchange()            .flatMapMany { response ->                LOGGER.info("Received cities..")                response.bodyToFlux<Long>()            }}

我正在使用 Spring 優(yōu)秀的 WebClient 庫(kù)進(jìn)行遠(yuǎn)程調(diào)用并獲得 Project Reactor Mono <ClientResponse> 類型的響應(yīng)，可以使用 flatMapMany 運(yùn)算符將其修改為 Flux<Long> 類型。

根據(jù)城市 ID，沿著同樣的路線獲取城市的詳情：

private fun getCityDetail(cityId: Long?): Mono<City> {    return webClient.get()            .uri("/cities/{id}", cityId!!)            .exchange()            .flatMap { response ->                val city: Mono<City> = response.bodyToMono()                LOGGER.info("Received city..")                city            }}

在這里，Project Reactor Mono<ClientResponse> 類型正在使用 flatMap 運(yùn)算符轉(zhuǎn)換為 Mono<City> 類型。

以及從中獲取 cityIds，這是 City 的代碼：

val cityIdsFlux: Flux<Long> = getCityIds()val citiesFlux: Flux<City> = cityIdsFlux        .flatMap { this.getCityDetail(it) }return citiesFlux

這非常具有表現(xiàn)力 - 對(duì)比基于回調(diào)的方法的混亂和基于 Reactive Streams 的方法的簡(jiǎn)單性。

到此，關(guān)于“怎么理解Java的回調(diào)與反應(yīng)模式”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)砀鄬?shí)用的文章！