SSE流式响应像chatGPT聊天一样

SSE是什么

SSE（Server-Sent Events）是一种基于 HTTP 协议的技术，允许服务器向客户端主动推送消息，而不是依赖客户端轮询获取数据。它在单向的情况下工作非常有效，即服务器持续发送数据更新给客户端，而客户端只接收数据，不向服务器发送额外请求。

SSE 的特点：

单向推送：服务器向客户端发送数据。客户端只需最初发起一次连接，服务器之后可以通过同一个连接不断发送数据。

长连接：SSE 使用了一个持久的 HTTP 连接，服务器在连接上持续发送事件，不会在每次更新时重新建立连接，这种方式减少了网络开销。

自动重连：如果连接断开，SSE 会自动尝试重连，保证数据流的持续性。

文本格式：SSE 发送的数据是文本格式，可以是简单的文本，也可以是 JSON 等其他结构化数据。

前端实现


<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Reactive SSE</title>
</head>
<body>
    <h1>Server-Sent Events: User Data</h1>
    <ul id="user-list"></ul>

    <script>
        const eventSource = new EventSource("/stream-users");

        eventSource.onmessage = function(event) {
            const user = JSON.parse(event.data);  // 将 JSON 数据转换为对象
            const userItem = document.createElement("li");
            userItem.textContent = `Name: ${user.name}, Email: ${user.email}`;
            document.getElementById("user-list").appendChild(userItem);
        };

        eventSource.onerror = function(event) {
            console.error("Error occurred:", event);
        };
    </script>
</body>
</html>

java 后端实现

    
        io.projectreactor
        reactor-core
        test

@RestController
public class UserController {

    @Autowired
    private UserRepository userRepository;

    // SSE 实现：每 2 秒发送一次用户数据
    @GetMapping(value = "/stream-users", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<User> streamUsers() {
        return userRepository.findAll()
                             .delayElements(Duration.ofSeconds(2));  // 每2秒发送一个用户数据
    }
}

public interface UserRepository extends ReactiveMongoRepository<User, String> {
    // 查询所有用户
    Flux<User> findAll();
}

前端示例2

  initSSES: async function(){
    if ('EventSource' in window && Ext.fetchEventSource) {
      await Ext.fetchEventSource(Ext.baseURL+"/xml/fileStatus", {
        method: 'GET',
        headers: { 'Authorization': 'Bearer '+Ext.util.Cookies.get("UserToken") },
        onopen(e){ console.log('onopen', e) },
        onclose(e){ console.log('onclose', e) },
        onmessage(e) {
          console.log('onmessage', e);
          const grid = Ext.ComponentQuery.query('#file-list-download')
          if (!grid || !grid.length){
            return
          }
          try {
            const message = JSON.parse(e.data)
            var record = grid[0].getStore().getById(message.fileId);
            if(record){
              record.set('status', message.status)
              record.set('path', message.path)
              record.set('errorReason', message.errorReason)
              record.commit()
            }else{
              grid[0].getStore().reload()
            }
          } catch (error) {
            grid[0].getStore().reload()
          }
        },
        onerror(e) { console.log('onerror', e); },
      });
    }
  },

{{userData.name}}已认证