# qianfan-sse-demo

**Repository Path**: xshuai/qianfan-sse-demo

## Basic Information

- **Project Name**: qianfan-sse-demo
- **Description**: 🍬基于https://gitee.com/codinginn/chatgpg-sse-demo-springboot-vue改动
本项目用SpringBoot和vue实现了仿各大大模型体验网页的打字机效果， 克隆项目到本地可以直接运行。主要应用的技术有:SpringBoot、Vue、Reactive、WebFlux、EventSource等，学习和练手的好项目。
- **Primary Language**: Java
- **License**: MulanPSL-2.0
- **Default Branch**: master
- **Homepage**: https://www.ydxiaoshuai.cn
- **GVP Project**: No

## Statistics

- **Stars**: 19
- **Forks**: 6
- **Created**: 2024-02-02
- **Last Updated**: 2025-02-25

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

本项目用SpringBoot和vue实现了仿各大大模型体验网页的打字机效果，
克隆项目到本地可以直接运行。主要应用的技术有:SpringBoot、Vue、Reactive、WebFlux、EventSource等，学习和练手的好项目。

![演示效果](img/%E7%99%BE%E5%BA%A6%E5%8D%83%E5%B8%86%E6%BC%94%E7%A4%BASSE.gif)

chat-sse-springboot 后端项目

chat-sse-vue        前端项目


使用的原始项目：[https://gitee.com/codinginn/chatgpg-sse-demo-springboot-vue](https://gitee.com/codinginn/chatgpg-sse-demo-springboot-vue)，感谢大佬开源

准备好了吗，<font color="red">let's get it!</font> 

## 项目运行

SpringBoot安装依赖，并设置对应的百度千帆APIKEY、SECRETKEY即可：

前端项目采用vue3实现，

在项目中,使用如下命令运行项目，即可运行前端:

```bash
yarn install
yarn serve
```

## 在Spring Boot中实现SSE

### 设置SSE

在SpringBoot中实现SSE并不复杂。其核心在于使用Spring框架的`webflux`。使用webflux类能够创建一个持久的连接，使服务器能够向客户端发送多个事件，而无需每次都建立新的连接。

先在pom中引入相关依赖：

```xml
<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
</dependency>

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

然后需要在Controller中创建一个方法来返回`Flux`实例。这个方法将被映射到特定的URL，客户端将使用这个URL来接收事件。

```java
@RestController
public class QianFanController {

    //用于流式请求第三方的实现类
    @Resource
    GptQianFanServiceImpl qianFanService;

    //通过stream返回流式数据
    @GetMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<QianFanResponseBean>> getChat(@RequestParam("content")String content) {
        //实现类发送消息并获取返回结果
        return qianFanService.doQianFanStream(content)
                //进行结果的封装，再返回给前端
                .map(aiAnswerDTO -> ServerSentEvent.<QianFanResponseBean>builder()
                        .data(aiAnswerDTO)
                        .build()
                )
                //发生异常时发送空对象
                .onErrorResume(e -> Flux.empty());
    }
```

### 处理连接和事件

一旦有客户端连接到这个URL，可以通过调用`GptQianFanServiceImpl`实例的`doQianFanStream`方法来发送事件。这些事件可以是简单的字符串消息，也可以是更复杂的数据结构，如JSON对象。记住，SSE的设计初衷是轻量级和简单，所以你发送的每个事件都应当是独立的和自包含的。

`GptQianFanServiceImpl`的实现方式如下，也是SpringBoot后端实现的重点

```java
/**
 * @Author 小帅丶
 * @Description 实现类
 * @Date  2024/2/2  16:00:33
 * @Param
 * @return
 **/
@Slf4j
@Service
public class GptQianFanServiceImpl {
    //webflux的client
    private WebClient webClient;

    @Value("${bdqfllm.API_KEY}")
    private String API_KEY;

    @Value("${bdqfllm.SECRET_KEY}")
    private String SECRET_KEY;

    @Value("${bdqfllm.API_HOST}")
    private String API_HOST;

    @Value("${bdqfllm.API_URI}")
    private String API_URI;

    @Value("${bdqfllm.TOKEN_URL}")
    private String TOKEN_URL;

    //用于读取第三方的返回结果
    private ObjectMapper objectMapper = new ObjectMapper();

    @PostConstruct
    public void postConstruct() {
        //创建webflux的client
        this.webClient = WebClient.builder()
                //填写对应的api地址
                .baseUrl(API_HOST)
                //设置默认请求类型
                .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
                .build();
    }

    //请求stream的主题
    public Flux<QianFanResponseBean> doQianFanStream(String content) {
        String requestURI = API_URI.replace("ACCESS_TOKEN",getAccessToken());
        //组装请求参数
        QianFanChatBean requestBean =  getRequestData(content);
        //构建请求json
        String paramJson = JSONUtil.toJsonStr(requestBean);;
        //使用webClient发送消息
        return this.webClient.post()
                //请求uri
                .uri(requestURI)
                //设置流式响应
                .header(HttpHeaders.CONTENT_TYPE, MediaType.TEXT_EVENT_STREAM_VALUE)
                .contentType(MediaType.APPLICATION_JSON)
                .acceptCharset(Charset.defaultCharset())
                .body(BodyInserters.fromValue(paramJson))
                .retrieve()
                .bodyToFlux(String.class)
                //接收到消息的处理方法
                .flatMap(result -> handleWebClientResponse(result));

    }

    private Flux<QianFanResponseBean> handleWebClientResponse(String resp) {
        log.info("百度千帆返回:======>{}",resp);
        QianFanResponseBean result = JSON.parseObject(resp, QianFanResponseBean.class);
        //is_end=true结束标识
        if (result.isIs_end()){
            return Flux.empty();
        }
        try {
            //返回获得的结果
            return Flux.just(result);
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        }
    }

    public  String getAccessToken(){
        String REQ_URL = TOKEN_URL.replace("API_KEY",API_KEY)
                .replace("SECRET_KEY",SECRET_KEY);
        String result = HttpRequest.get(REQ_URL).execute().body();
        QFAccessTokenBean tokenBean = JSON.parseObject(result,QFAccessTokenBean.class);
        return tokenBean.getAccess_token();
    }

    public static QianFanChatBean getRequestData(String content) {
        QianFanChatBean requestBean = new QianFanChatBean();
        List<QianFanChatBean.QFMessage> messages = new ArrayList<>();
        QianFanChatBean.QFMessage message = new QianFanChatBean.QFMessage();
        message.setRole(QFConts.QFRole.USER);
        message.setContent(content);
        messages.add(message);
        requestBean.setMessages(messages);
        requestBean.setStream(true);
        return requestBean;
    }
```


### 处理异常和断开连接

在使用SSE时，处理异常和断开连接也非常重要。确保在客户端断开连接或发生异常时，正确地关闭`webflux`实例。这有助于避免资源泄露和其他潜在问题。

### 一些实用的提示

- **超时管理**：SSE连接可能因为超时而被关闭。确保妥善处理这种情况。
- **错误处理**：适当地处理可能发生的异常，如网络问题或客户端断开连接。
- **资源清理**：在连接结束时清理资源，确保应用的健康和性能。

**至此，我们就完成了SpringBoot的SSE后端开发。**

## Vue前端对SSE的处理

### 在Vue中接收SSE

在Vue应用中接收SSE消息是相对直截了当的。需要做的基本上就是在Vue组件中创建一个新的`EventSource`实例，并指向你的SpringBoot应用中设置的SSE URL,本文使用EventSource作为示例，也可以选择axios或@microsoft/fetch-event-source发送post请求的SSE请求，使用另外两种的好处是可以控制header，携带token信息，以便于控制权限。

```js
        this.eventSource = new EventSource('http://127.0.0.1:8080/chat?content='+this.inputText);
```

一旦建立了连接，就可以定义各种事件监听器来处理从服务器接收到的消息。在Vue中，这通常涉及到更新组件的数据属性，这些属性又通过Vue的响应式系统自动更新UI。

```js
sendSSEMessage() {
      // 只有当eventSource不存在时才创建新的EventSource连接
      if (!this.eventSource) {
        this.messages.push({text: this.inputText, isMine: true});
        this.messages.push({text: "", isMine: false});

        // 创建新的EventSource连接
        this.eventSource = new EventSource('http://127.0.0.1:8080/chat?content='+this.inputText);

        // 设置消息接收的回调函数
        this.eventSource.onmessage = (event) => {
          const data = JSON.parse(event.data);
          this.messages[this.messages.length - 1].text += data.result;
        };

        // 可选：监听错误事件，以便在出现问题时能够重新连接或处理错误
        this.eventSource.onerror = (event) => {
          console.error("EventSource failed:", event);
          this.eventSource.close(); // 关闭出错的连接
          this.eventSource = null; // 重置eventSource变量，允许重建连接
        };
      }
    }
```

### 保证流畅的用户体验

当处理实时数据时，保证一个流畅且不中断的用户体验至关重要。在Vue中，这意味着需要确保UI的更新是平滑和高效的。幸运的是，Vue的响应式系统会处理大部分重活，但你仍需要注意不要进行不必要的大规模DOM操作或数据处理。

### 异常处理和重连

处理连接中断或其他异常也是至关重要的。你可能需要在失去连接时尝试重新连接，或者至少提醒用户当前的连接状态。这可以通过监听`EventSource`的错误事件并采取适当的行动来实现。

### 小结

将SSE与Vue结合使用，可以为用户提供一个富有动态性和实时性的界面。无论是实时消息、通知，还是实时数据流，SSE都能让Vue应用更加生动和实用。