哈尔滨建设网站制作,宁波本地网站排行,wordpress 和shopify,自己做网站租服务器响应式编程#xff08;Reactive Programming#xff09;是一种编程范式#xff0c;旨在处理异步数据流和事件流。它通过使用观察者模式和函数式编程的概念#xff0c;将数据流和事件流抽象为可观察的序列#xff0c;然后通过操作这些序列来实现各种功能。
在响应式编程中…响应式编程Reactive Programming是一种编程范式旨在处理异步数据流和事件流。它通过使用观察者模式和函数式编程的概念将数据流和事件流抽象为可观察的序列然后通过操作这些序列来实现各种功能。
在响应式编程中数据流和事件流被视为连续的时间序列可以通过操作符来转换、过滤和组合它们。这种编程范式的主要优势是它可以简化异步编程并提供一种声明式的方式来处理数据流和事件流。它还可以提高代码的可读性和可维护性因为它将复杂的异步逻辑封装在操作符中使得代码更易于理解和修改。
响应式编程可以应用于各种领域包括前端开发、后端开发、移动开发等。在前端开发中响应式编程可以用于处理用户界面的事件流和数据流使得界面能够动态地响应用户的操作。在后端开发中响应式编程可以用于处理大量的异步请求和数据流提高系统的吞吐量和响应速度。
常见的响应式编程框架包括RxJava、RxJS、ReactiveX等。这些框架提供了一系列的操作符和工具用于处理数据流和事件流并提供了一种简洁而强大的方式来处理异步编程。
当前响应式编程的典型例子莫过于最近炙手可热的ChatGPT的流式输出了。因为ChatGPT请求响应时间较长如果采用传统的一直等待全部数据就绪用户恐怕早就跑光了而响应式方式则不需要等待所有数据就绪而只需要有部分数据就绪即可输出从而极大地提升了用户体验。下面以此为例来说明实现这种效果的原理开发语言Java。
先来看看上文中提到的的三个响应式编程框架RxJava、RxJS和ReactiveX。它们是三个相关的概念同时也是不同平台上的实现。
RxJavaRxJava是ReactiveX在Java平台上的实现它提供了一套丰富的API和操作符用于处理异步和事件驱动的编程。RxJava是基于观察者模式和迭代器模式的可以用于处理数据流、事件流和异步任务等。RxJSRxJS是ReactiveX在JavaScript平台上的实现它提供了类似于RxJava的API和操作符用于处理异步和事件驱动的编程。RxJS可以在浏览器端和Node.js环境中使用可以处理DOM事件、AJAX请求、定时器等。ReactiveXReactiveX是一个跨平台的响应式编程库它提供了一套统一的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();
FlowableString response webClient.get().uri(https://example.com).retrieve().bodyToFlux(String.class).toFlowable();同样WebClient也可以与Reactor的Flux一起使用通过bodyToFlux()方法将响应转换为Flux流从而实现对响应的处理和操作。
WebClient webClient WebClient.create();
FluxString 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。
FluxInteger flux Flux.just(1, 2, 3);
MonoInteger mono flux.single(); Mono可以被转换成Flux。
MonoInteger mono Mono.just(1);
FluxInteger 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 PublisherString 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
