大家好，我是程序媛雪儿，有两天没和大家唠唠技术了，今天我们聊聊AI流式调用，这块我也是总结鱼皮哥AI答题应用平台项目中的技术，大家看了有兴趣可以学习一下，实操体验一下怎么应用哦~   现在AI已经在各大软件中广泛应用，你们有没有想过如何在AI应用中实时处理数据流？今天咱们聊聊怎么用SSE Rxjava处理实时数据流。 一、SSE是什么 SSE（后端主动推送给前端） 前端发请求并和后端建立连接，后端实时推动数据给前端 SSE的重要特点 单向通信：SSE只支持服务器向客户端的单向通信 文本格式：SSE使用纯文本格式传输数据，HTTP响应的text/event-stream 保持连接：SSE会保持一个持久的HTTP连接，实现服务器向客户端推送数据 自动重连：如果连接中断，浏览器会尝试自动重连 为什么处理AI流式数据要用SSE？ AI对话是服务器单向给客户端流式传输数据，用SSE更加 简单：不需要WebSocket那么复杂，基本的HTTP和JavaScript就搞定。 实时更新：长连接，随时获取最新数据。 轻量：适合频繁更新的小数据量。 二、RxJava是什么 RxJava是一个基于事件驱动的、利用可观测序列来实现异步编程的类库 1、事件驱动 事件可以是任何事情。比如用户的点击操作、网络请求的结果、文件的读写等 2、可观测序列 可观测序列指一系列按照时间序列发出的数据项，可以被观察处理 RxJava的核心知识点 观察者模式 RxJava是基于观察者模式实现的。 观察者：观测数据流 observer 被观察者：实时传输数据流 observable和flowable observable适合处理相对较小的、可控的、不会产生大量数据的场景，不具备背压能力 flowable具备背压能力。也就是说，如果生产数据过快，超过了大多数数据消费者速度，flowable提供了多种背压策略来处理这种情况，保证大量数据仍然能稳定 建立订阅关系 被观察者.subscribe(观察者) 三、后端Rxjava流式调用的demo    // region    // 生成AI题目流式生成    @GetMapping("/ai_generate/sse")    public SseEmitter aiGenerateSSE(AiGenerateRequest aiGenerateRequest){         // 获取应用信息        // 建立SSE连接对象，0表示永不超时        SseEmitter sseEmitter = new SseEmitter(0L);        // AI生成（调用AI流式接口），SSE流式返回        Flowable<ModelData> modelDataFlowable = aiManager.doStreamRequest(GENERATE_QUESTION_SYSTEM_MESSAGE, userMessage, null);         // 截取流式数据进行数据处理后返回给前端        modelDataFlowable                // 指定观察者的线程池                .observeOn(Schedulers.io())                // 先获取数据                .map(modelData -> modelData.getChoices().get(0).getDelta().getContent())                // 先处理数据把没用的空格都去掉                .map(message -> message.replaceAll("\\s",""))                .filter(StrUtil::isNotBlank)                .flatMap(message ->{                    List<Character> characterList = new ArrayList<>();                    for (char c : message.toCharArray()) {                        characterList.add(c);                    }                    return Flowable.fromIterable(characterList);                })                .doOnNext(c -> {                    // 按照业务需要处理数据                })                .doOnError((e) -> log.error("异常处理"))                .doOnComplete(()->{                    sseEmitter.complete();                })                .subscribe();        return sseEmitter;    }    // endregion 04 四、前端使用sse开启连接，获取数据 /** * 提交流式生成题目，一个一个的生成题目 */const handleSSESubmit = async () => {  // 创建SSE请求  const eventSource = new EventSource(    // 手动填写完整的后端地址    "http://localhost:8101/ai_generate/sse"  );  let first = true;  // 接收消息  eventSource.onmessage = function(event) {    console.log(event.data);    if(first){      console.log('第一次连接');      first = !first;    }    console.log('传输数据',event.data);  };  // 报错或连接关闭时触发  eventSource.onerror = function(event) {    // 关闭SSE连接    if(event.eventPhase === EventSource.CLOSED){      console.log('关闭连接');      eventSource.close();    }  };  // 连接打开时触发  eventSource.onopen = function(event) {    console.log('连接成功');  };};     基本上就是后端用Rxjava框架观察处理数据，处理成前端需要的形式传给前端，前端用SSE的方式接收数据，进行实时的数据展示。是不是很简单​？那今天雪儿的分享就结束啦，也欢迎各位宝宝关注我的微信公众号，程序媛雪儿，雪儿基本上天天都会分享技术碎片，我们一起学习，一起进步，加油哦~#知识碎片# #学习总结#