响应式编程（Reactive Programming）是一种编程范式，旨在处理异步数据流和事件流。它通过使用观察者模式和函数式编程的概念，将数据流和事件流抽象为可观察的序列，然后通过操作这些序列来实现各种功能。
在响应式编程中，数据流和事件流被视为连续的时间序列，可以通过操作符来转换、过滤和组合它们。这种编程范式的主要优势是它可以简化异步编程，并提供一种声明式的方式来处理数据流和事件流。它还可以提高代码的可读性和可维护性，因为它将复杂的异步逻辑封装在操作符中，使得代码更易于理解和修改。
响应式编程可以应用于各种领域，包括前端开发、后端开发、移动开发等。在前端开发中，响应式编程可以用于处理用户界面的事件流和数据流，使得界面能够动态地响应用户的操作。在后端开发中，响应式编程可以用于处理大量的异步请求和数据流，提高系统的吞吐量和响应速度。
常见的响应式编程框架包括RxJava、RxJS、ReactiveX等。这些框架提供了一系列的操作符和工具，用于处理数据流和事件流，并提供了一种简洁而强大的方式来处理异步编程。
当前响应式编程的典型例子莫过于最近炙手可热的ChatGPT的流式输出了。因为ChatGPT请求响应时间较长，如果采用传统的一直等待全部数据就绪，用户恐怕早就跑光了，而响应式方式则不需要等待所有数据就绪，而只需要有部分数据就绪即可输出，从而极大地提升了用户体验。下面以此为例，来说明实现这种效果的原理（开发语言Java）。
先来看看上文中提到的的三个响应式编程框架：RxJava、RxJS和ReactiveX。它们是三个相关的概念，同时也是不同平台上的实现。
- RxJava：RxJava是ReactiveX在Java平台上的实现，它提供了一套丰富的API和操作符，用于处理异步和事件驱动的编程。RxJava是基于观察者模式和迭代器模式的，可以用于处理数据流、事件流和异步任务等。
- RxJS：RxJS是ReactiveX在JavaScript平台上的实现，它提供了类似于RxJava的API和操作符，用于处理异步和事件驱动的编程。RxJS可以在浏览器端和Node.js环境中使用，可以处理DOM事件、AJAX请求、定时器等。
- ReactiveX：ReactiveX是一个跨平台的响应式编程库，它提供了一套统一的API和操作符，用于处理异步和事件驱动的编程。ReactiveX的目标是提供一种通用的编程模型，使得开发者可以在不同的平台和语言中共享代码和思想。
在Springboot中，另有WebFlux模块可供使用，同时它也可以跟上面的模块一起使用。说起Flux，这里也会涉及到另一个概念：Flowable。其实Flowable和Flux都是响应式流的实现，它们有以下关系：
- Flowable是RxJava的一部分，而Flux是Reactor的一部分。RxJava是一个用于Java的响应式编程库，而Reactor是一个用于Java的响应式编程框架。
- Flowable是RxJava中的一个类，它实现了Reactive-Streams规范，提供了对背压（backpressure）的支持。Flowable可以处理异步和并发的数据流，并且可以控制数据流的速率，以避免生产者和消费者之间的不匹配。
- Flux是Reactor中的一个类，它也实现了Reactive-Streams规范，提供了类似的功能。Flux可以处理异步和并发的数据流，并且可以控制数据流的速率。
- Flowable和Flux都提供了一系列的操作符，可以对数据流进行转换、过滤、映射等操作。这些操作符可以帮助开发者处理和操作数据流，使代码更加简洁和可读。
跟tRxJava和Reactor密切相关的开发库之一是WebClien。WebClient是一个用于发送HTTP请求的非阻塞的响应式客户端，它是Reactor项目的一部分。
WebClient提供了一种简洁、灵活和可组合的方式来发送HTTP请求，并处理响应。它可以与RxJava和Reactor的异步和响应式编程模型无缝集成，使得在响应式应用程序中处理HTTP请求变得更加方便和高效。
WebClient可以与RxJava的Flowable一起使用，通过toFlowable()方法将响应转换为Flowable流，从而实现对响应的处理和操作。
WebClient webClient = WebClient.create();
Flowable<String> response = webClient.get()
.uri("https://example.com")
.retrieve()
.bodyToFlux(String.class)
.toFlowable();
同样，WebClient也可以与Reactor的Flux一起使用，通过bodyToFlux()方法将响应转换为Flux流，从而实现对响应的处理和操作。
WebClient webClient = WebClient.create();
Flux<String> response = webClient.get()
.uri("https://example.com")
.retrieve()
.bodyToFlux(String.class);下面我们将关注点放在Reactor框架中，在Reactor中，不得不提另一个跟Flux相对的概念：Mono。Flux和Mono是Reactor框架中的两个关键类，它们都是用于处理响应式流的。
- Flux是一个表示0到N个元素的响应式流。它可以发出多个元素，并以异步的方式产生这些元素。Flux可以用于处理多个值的数据流，例如从数据库查询结果、文件读取等。
- Mono是一个表示0或1个元素的响应式流。它要么发出一个元素，要么不发出任何元素。Mono可以用于处理单个值的数据流，例如从缓存中获取数据、获取单个实体等。
- Flux和Mono之间有以下关系：
- Flux可以被转换成Mono。
Flux<Integer> flux = Flux.just(1, 2, 3);
Mono<Integer> mono = flux.single();-
- Mono可以被转换成Flux。
Mono<Integer> mono = Mono.just(1);
Flux<Integer> flux = mono.flux();Flux和Mono可以通过一系列的操作符进行转换、过滤、映射等操作，使得对响应式流的处理变得更加灵活和方便。它们是Reactor框架中的核心类，用于构建响应式应用程序。
webClient可以实现复杂的处理逻辑，比如异常处理：
webClient.get()
.uri(url)
.retrieve()
.onStatus(HttpStatus::is4xxClientError, response -> Mono.error(new CustomException("客户端错误")))
.onStatus(HttpStatus::is5xxServerError, response -> Mono.error(new CustomException("服务器错误")))
.bodyToMono(String.class)
.onErrorResume(throwable -> {
if (throwable instanceof WebClientResponseException) {
WebClientResponseException ex = (WebClientResponseException) throwable;
// 处理响应异常
} else {
// 处理其他异常
}
});在使用 Spring Boot 的 WebClient 时，bodyToMono 和 bodyToFlux 方法都可以用于将响应体转换为 Mono 或 Flux 对象。
bodyToMono 方法用于将响应体转换为 Mono 对象，适用于响应体只有一个元素的情况，例如返回一个 JSON 对象或者一个字符串。
bodyToFlux 方法用于将响应体转换为 Flux 对象，适用于响应体有多个元素的情况，例如返回一个 JSON 数组或者一个流式数据。
因此，当我们需要处理的响应体只有一个元素时，应该使用 bodyToMono 方法；当我们需要处理的响应体有多个元素时，应该使用 bodyToFlux 方法。
在 Reactor 中，Flux 流结束的实现原理是通过发送一个 onComplete 信号来通知订阅者流已经结束。当 Flux 流中的所有元素都被消费完毕时，会自动发送一个 onComplete 信号。
例如，当我们使用 Flux.range(1, 10) 创建一个包含 1 到 10 的整数序列的 Flux 流时，当订阅者订阅该流并消费完所有元素后，会自动发送一个 onComplete 信号来通知订阅者流已经结束。
在使用 Spring Boot 的 WebClient 时，当我们使用 bodyToFlux 方法将响应体转换为 Flux 对象时，如果响应体是一个流式数据，那么当流式数据传输完毕后，会自动发送一个 onComplete 信号来通知订阅者流已经结束。
webClient.get()
.uri(url)
.retrieve()
.bodyToFlux(String.class)
.doFinally(signalType -> {
if (signalType == SignalType.ON_COMPLETE) {
System.out.println("流已结束");
}
})
.subscribe();有了这些基础知识的准备，我们再来看看ChatGPT的响应结果样例。OpenAI的聊天接口是：
http://api.openai.com/v1/chat/completitions。
该接口接受这样的一个请求数据结构：ChatCompletionRequest。其中有个属性stream 可以设定是否采用流输出。默认false。
这个例子是非stream输出，输出格式为：ChatCompletionResponse
$ curl https://api.openai.com/v1/chat/completions -H 'Content-Type: application/json' -H "Authorization: Bearer sk-zDxkX0Na0e63B18c9c6bT3BlBkFJf3De387b398749c5bD1d" -d '{"model": "gpt-3.5-turbo","stream":"false","messages": [{"role": "user", "content": "Hello!"}]}'
{"id":"chatcmpl-7tywVQ4vSPzs8yuZy5FqvL0CX07W0","object":"chat.completion","created":1693576659,"model":"gpt-3.5-turbo-0613","choices":[{"index":0,"message":{"role":"assistant","content":"Hello这个例子是stream输出，输出结构体为：字符串格式的ChatCompletionResponse：
curl https://api.openai.com/v1/chat/completions -H 'Content-Type: application/json' -H "Authorization: Bearer sk-zDxkX0Na0e63B18c9c6bT3BlBkFJf3De387b398749c5bD1d" -d '{"model": "gpt-3.5-turbo","stream":"true","messages": [{"role": "user", "content": "Hello!"}]}'
比较stream和非stream的输出区别，有一下几点：
1.非stream 输出只有一条记录；stream 有若干条，取决于响应内容大小；
2. 非stream 输出包含消耗的tokens数量，stream 没有；
3. 非stream 输出结果是json格式的ChatCompletionResponse结构，stream 输出j格式类似：data:str(ChatCompletionResponse)，同时以data:[NONE]结尾；
结合上面的知识，我们就能实现上述功能：
public Publisher<String> generateChatCompletion(ChatCompletionRequest chatCompletionRequest) {
WebClient.ResponseSpec responseSpec = webClient.post()
.uri(this.apiUrl + "/chat/completions").header("Authorization", "Bearer " + this.apiKey)// .accept(MediaType.TEXT_EVENT_STREAM)
.bodyValue(chatCompletionRequest)
.retrieve();
if (chatCompletionRequest.getStream())
return
responseSpec.bodyToFlux(ChatCompletionResponse.class)
.onErrorResume(error -> {
// 异常处理逻辑
logger.error("bodyToFlux error: {}", error);
return Flux.empty();
})
.filter(response -> {
ChatMessage message = response.getChoices().get(0).getMessage();
if (message != null) {
String content = message.getContent();
return StringUtils.isNotEmpty(StringUtils.trim(content));
}
return false;
})
.mapNotNull(response -> {
try {
return objectMapper.writeValueAsString(response);
} catch (JsonProcessingException e) {
logger.error(e);
return null;
}
}).concatWithValues("[DONE]");
else
return
responseSpec.bodyToMono(ChatCompletionResponse.class)
.onErrorResume(error -> {
// 异常处理逻辑
logger.error("bodyToMono error: {}", error);
return Mono.empty();
}).mapNotNull(response -> {
try {
return objectMapper.writeValueAsString(response);
} catch (JsonProcessingException e) {
logger.error(e);
return null;
}
});
}Publisher是一个通用的概念，它代表一个发布者，可以发布数据或事件。在Spring WebFlux中，Flux和Mono都是Publisher的实现类。
试用地址：https://chatgpt-discount.zeabur.app
