使用Spring WebFlux + Websocket + WebClient 响应式请求openai 接口

导入依赖

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>2.0.32</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>

ChatGPTController

@Slf4j
public class MyWebSocketHandler1 implements WebSocketHandler {

    @Override
    public Mono<Void> handle(WebSocketSession session) {
        return session.receive()
                .map(webSocketMessage -> {
                    return webClient().post()
                            .accept(MediaType.TEXT_EVENT_STREAM)
                            .body(BodyInserters.fromValue(jsonObject(webSocketMessage.getPayloadAsText())))
                            .retrieve()
                            .bodyToFlux(String.class)
                            .map(s -> {
                                if (!Objects.equals(s, "[DONE]")) {
                                    JSONObject jo = JSON.parseObject(s).getJSONArray("choices").getJSONObject(0).getJSONObject("delta");
                                    String content = jo.getString("content");
                                    if(content != null){
                                        return session.textMessage(content);
                                    }
                                }
                                return session.textMessage("");
                            })
                            .onErrorResume(WebClientResponseException.class, ex -> Flux.just(session.textMessage(ex.getResponseBodyAsString())))
                            .doFinally(signalType -> log.info("完成"));
                })
                .flatMap(session::send)
                .then();
    }

    private JSONObject jsonObject(String content){
        JSONObject jsonObject = new JSONObject();
        JSONObject userMessage = new JSONObject();
        userMessage.put("role","user");
        userMessage.put("content",content);
        JSONArray jsonArray = new JSONArray();
        jsonArray.add(userMessage);
        jsonObject.put("model", "gpt-3.5-turbo-16k-0613"); //速度快，价格gao
        jsonObject.put("messages", jsonArray);
        jsonObject.put("stream", true);
        return jsonObject;
    }

    private WebClient webClient(){
        return  WebClient.builder()
                .clientConnector(new ReactorClientHttpConnector(
                        HttpClient.create().proxy(proxy -> proxy.type(ProxyProvider.Proxy.HTTP).host("127.0.0.1").port(1080))
                ))
                .defaultHeader("Content-Type", MediaType.APPLICATION_JSON_VALUE)
                .defaultHeader("Authorization", "Bearer ")
                .baseUrl("https://api.openai.com/v1/chat/completions")
                .build();
    }
}

WebConfig 配置

@Configuration
@EnableWebFlux
public class WebConfig implements WebFluxConfigurer {

    @Override
    public void addCorsMappings(CorsRegistry registry) {
        registry.addMapping("/**")
                .allowedHeaders("*")
                .allowedMethods("*")
                .allowedOriginPatterns("*")
                .allowCredentials(true);
    }

    @Bean
    public HandlerMapping handlerMapping() {
        Map<String, WebSocketHandler> map = new HashMap<>();
        map.put("/path/{roomId}", new MyWebSocketHandler1());
        int order = -1; // before annotated controllers
        return new SimpleUrlHandlerMapping(map, order);
    }

}