Java調(diào)用ChatGPT API并實(shí)現(xiàn)流式接收方式(Server-Sent Events,SSE)

更新時間：2025年08月19日 17:11:13 作者：陳震_

文章介紹如何在Java中通過OkHttp和SSE技術(shù)實(shí)現(xiàn)流式獲取ChatGPT響應(yīng),解決傳統(tǒng)HTTP阻塞問題,提升用戶體驗(yàn),需配置stream參數(shù),利用SseEmitter封裝后端推送,前端使用EventSourcePolyfill插件處理Token,同時注意資源管理和避免換行符干擾

簡介

用過 ChatGPT 的伙伴應(yīng)該想過自己通過調(diào)用ChatGPT官網(wǎng)提供的接口來實(shí)現(xiàn)一個自己的問答機(jī)器人，但是在調(diào)用的時候發(fā)現(xiàn)，請求總是以傳統(tǒng)的HTTP請求/響應(yīng)模式進(jìn)行，這意味著我們沒發(fā)送一個請求后需要等待 ChatGPT 服務(wù)器返回完整的響應(yīng)。這種方式在生成文本時并不不是我們理想的，因?yàn)橛脩趔w驗(yàn)不夠流暢。

為了提供更好的用戶體驗(yàn)，我們可以使用Server-Sent Events（SSE）技術(shù)來實(shí)現(xiàn)流式接收。這樣，當(dāng)ChatGPT 服務(wù)器可以在生成響應(yīng)的同時逐步將內(nèi)容推送給我們，我們在通過 SSE 流式推送到前端頁面，讓用戶能夠?qū)崟r看到生成的內(nèi)容。我將詳細(xì)介紹如何在Java中實(shí)現(xiàn)這一功能。

OkHttp 流式獲取 GPT 響應(yīng)

其實(shí)市面上已經(jīng)有很多現(xiàn)成的框架支持，但我們這里使用 okHttp 這個輕量級的HTTP客戶端庫來實(shí)現(xiàn)。

需要先引用相關(guān)maven：

    <dependency>
        <groupId>com.squareup.okhttp3</groupId>
        <artifactId>okhttp</artifactId>
    </dependency>
    <dependency>
        <groupId>com.squareup.okhttp3</groupId>
        <artifactId>okhttp-sse</artifactId>
    </dependency>

構(gòu)建請求體，必須加上參數(shù) stream 值為true

 //構(gòu)建發(fā)送內(nèi)容
        String messageStr = StrUtil.format(prompt, params);

        // 創(chuàng)建一個Message對象，該對象表示一個消息，并設(shè)置其屬性
        Message message = new Message(Message.Role.USER.getRole(), messageStr);
        // 創(chuàng)建一個ChatCompletion對象，表示聊天完成請求，并將剛創(chuàng)建的消息添加到其中
        ChatCompletionRequest request = ChatCompletionRequest.builder()
                .model(ChatCompletionRequest.Model.GPT_3_5_TURBO.getName())
                .messages(Arrays.asList(message))
                .stream(true)
                .build();

       
// 定義see接口
Request request = new Request.Builder().url("https://api.openai.com/v1/chat/completions")
        .header("Authorization","xxx")
        .post(okhttp3.RequestBody.create(okhttp3.MediaType.parse("application/json; charset=utf-8"),param.toJSONString()))
        .build();
OkHttpClient okHttpClient = new OkHttpClient.Builder()
        .connectTimeout(10, TimeUnit.MINUTES)
        .readTimeout(10, TimeUnit.MINUTES)//這邊需要將超時顯示設(shè)置長一點(diǎn)，不然剛連上就斷開，之前以為調(diào)用方式錯誤被坑了半天
        .build();

// 實(shí)例化EventSource，注冊EventSource監(jiān)聽器
RealEventSource realEventSource = new RealEventSource(request, new EventSourceListener() {

    @Override
    public void onOpen(EventSource eventSource, Response response) {
        log.info("onOpen");
    }

    @SneakyThrows
    @Override
    public void onEvent(EventSource eventSource, String id, String type, String data) {
		// log.info("onEvent");
		// 在實(shí)際應(yīng)用中，你可以在這里將數(shù)據(jù)推送給前端
        log.info(data);//請求到的數(shù)據(jù)
    
    }

    @Override
    public void onClosed(EventSource eventSource) {
        log.info("onClosed");
//                emitter.complete();
    }

    @Override
    public void onFailure(EventSource eventSource, Throwable t, Response response) {
        log.error("onFailure 出現(xiàn)異常，response={}", response, t);//這邊可以監(jiān)聽并重新打開
//                emitter.complete();
    }
});
realEventSource.connect(okHttpClient);//真正開始請求的一步

通過 SSE 流式推送前端

sse(Server Sent Event)，直譯為服務(wù)器發(fā)送事件，顧名思義，也就是客戶端可以獲取到服務(wù)器發(fā)送的事件

我們常見的 http 交互方式是客戶端發(fā)起請求，服務(wù)端響應(yīng)，然后一次請求完畢；但是在 sse 的場景下，客戶端發(fā)起請求，連接一直保持，服務(wù)端有數(shù)據(jù)就可以返回?cái)?shù)據(jù)給客戶端，這個返回可以是多次間隔的方式

原理是先建立鏈接，然后不斷發(fā)消息就可以

我們利用 springboot 封裝的 SseEmitter 來完成推送，需要用到以下依賴：

<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
    <groupId>cn.hutool</groupId>
    <artifactId>hutool-all</artifactId>
    <version>5.7.16</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-test</artifactId>
</dependency>

后端代碼

消息實(shí)體

其中客戶端 ID 是每個 SSE 鏈接的唯一標(biāo)識，拿到 ID 可以精準(zhǔn)的給唯一的用戶推送消息，消息通過字符串的方式進(jìn)行傳遞

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * 消息體
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class MessageVo {
    /**
     * 客戶端id
     */
    private String clientId;
    /**
     * 傳輸數(shù)據(jù)體(json)
     */
    private String data;
}

接口

import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;


public interface SseEmitterService {
    /**
     * 創(chuàng)建連接
     *
     * @param clientId 客戶端ID
     */
    SseEmitter createConnect(String clientId);

    /**
     * 根據(jù)客戶端id獲取SseEmitter對象
     *
     * @param clientId 客戶端ID
     */
    SseEmitter getSseEmitterByClientId(String clientId);

    /**
     * 發(fā)送消息給所有客戶端
     *
     * @param msg 消息內(nèi)容
     */
    void sendMessageToAllClient(String msg);

    /**
     * 給指定客戶端發(fā)送消息
     *
     * @param clientId 客戶端ID
     * @param msg      消息內(nèi)容
     */
    void sendMessageToOneClient(String clientId, String msg);

    /**
     * 關(guān)閉連接
     *
     * @param clientId 客戶端ID
     */
    void closeConnect(String clientId);
}

接口實(shí)現(xiàn)

@Slf4j
@Service
public class SseEmitterServiceImpl implements SseEmitterService {
    /**
     * 容器，保存連接，用于輸出返回 ;可使用其他方法實(shí)現(xiàn)
     */
    private static final Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();


    /**
     * 根據(jù)客戶端id獲取SseEmitter對象
     *
     * @param clientId 客戶端ID
     */
    @Override
    public SseEmitter getSseEmitterByClientId(String clientId) {
        return sseCache.get(clientId);
    }


    /**
     * 創(chuàng)建連接
     *
     * @param clientId 客戶端ID
     */
    @Override
    public SseEmitter createConnect(String clientId) {
        // 設(shè)置超時時間，0表示不過期。默認(rèn)30秒，超過時間未完成會拋出異常：AsyncRequestTimeoutException
        SseEmitter sseEmitter = new SseEmitter(0L);
        // 是否需要給客戶端推送ID
        if (StrUtil.isBlank(clientId)) {
            clientId = IdUtil.simpleUUID();
        }
        // 注冊回調(diào)
        sseEmitter.onCompletion(completionCallBack(clientId));     // 長鏈接完成后回調(diào)接口(即關(guān)閉連接時調(diào)用)
        sseEmitter.onTimeout(timeoutCallBack(clientId));        // 連接超時回調(diào)
        sseEmitter.onError(errorCallBack(clientId));          // 推送消息異常時，回調(diào)方法
        sseCache.put(clientId, sseEmitter);
        log.info("創(chuàng)建新的sse連接，當(dāng)前用戶：{}    累計(jì)用戶:{}", clientId, sseCache.size());
        try {
            // 注冊成功返回用戶信息
            sseEmitter.send(SseEmitter.event().id(String.valueOf(HttpStatus.HTTP_CREATED)).data(clientId, MediaType.APPLICATION_JSON));
        } catch (IOException e) {
            log.error("創(chuàng)建長鏈接異常，客戶端ID:{}   異常信息:{}", clientId, e.getMessage());
        }
        return sseEmitter;
    }

    /**
     * 發(fā)送消息給所有客戶端
     *
     * @param msg 消息內(nèi)容
     */
    @Override
    public void sendMessageToAllClient(String msg) {
        if (MapUtil.isEmpty(sseCache)) {
            return;
        }
        // 判斷發(fā)送的消息是否為空

        for (Map.Entry<String, SseEmitter> entry : sseCache.entrySet()) {
            MessageVo messageVo = new MessageVo();
            messageVo.setClientId(entry.getKey());
            messageVo.setData(msg);
            sendMsgToClientByClientId(entry.getKey(), messageVo, entry.getValue());
        }

    }

    /**
     * 給指定客戶端發(fā)送消息
     *
     * @param clientId 客戶端ID
     * @param msg      消息內(nèi)容
     */
    @Override
    public void sendMessageToOneClient(String clientId, String msg) {
        MessageVo messageVo = new MessageVo(clientId, msg);
        sendMsgToClientByClientId(clientId, messageVo, sseCache.get(clientId));
    }

    /**
     * 關(guān)閉連接
     *
     * @param clientId 客戶端ID
     */
    @Override
    public void closeConnect(String clientId) {
        SseEmitter sseEmitter = sseCache.get(clientId);
        if (sseEmitter != null) {
            sseEmitter.complete();
            removeUser(clientId);
        }
    }

    /**
     * 推送消息到客戶端
     * 此處做了推送失敗后，重試推送機(jī)制，可根據(jù)自己業(yè)務(wù)進(jìn)行修改
     *
     * @param clientId  客戶端ID
     * @param messageVo 推送信息，此處結(jié)合具體業(yè)務(wù)，定義自己的返回值即可
     **/
    private void sendMsgToClientByClientId(String clientId, MessageVo messageVo, SseEmitter sseEmitter) {
        if (sseEmitter == null) {
            log.error("推送消息失敗：客戶端{(lán)}未創(chuàng)建長鏈接,失敗消息:{}",
                    clientId, messageVo.toString());
            return;
        }
        SseEmitter.SseEventBuilder sendData = SseEmitter.event().id(String.valueOf(HttpStatus.HTTP_OK))
                .data(messageVo, MediaType.APPLICATION_JSON);
        try {
            sseEmitter.send(sendData);
        } catch (IOException e) {
            // 推送消息失敗，記錄錯誤日志，進(jìn)行重推
            log.error("推送消息失敗：{},嘗試進(jìn)行重推", messageVo.toString());
            boolean isSuccess = true;
            // 推送消息失敗后，每隔10s推送一次，推送5次
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(10000);
                    sseEmitter = sseCache.get(clientId);
                    if (sseEmitter == null) {
                        log.error("{}的第{}次消息重推失敗，未創(chuàng)建長鏈接", clientId, i + 1);
                        continue;
                    }
                    sseEmitter.send(sendData);
                } catch (Exception ex) {
                    log.error("{}的第{}次消息重推失敗", clientId, i + 1, ex);
                    continue;
                }
                log.info("{}的第{}次消息重推成功,{}", clientId, i + 1, messageVo.toString());
                return;
            }
        }
    }


    /**
     * 長鏈接完成后回調(diào)接口(即關(guān)閉連接時調(diào)用)
     *
     * @param clientId 客戶端ID
     **/
    private Runnable completionCallBack(String clientId) {
        return () -> {
            log.info("結(jié)束連接：{}", clientId);
            removeUser(clientId);
        };
    }

    /**
     * 連接超時時調(diào)用
     *
     * @param clientId 客戶端ID
     **/
    private Runnable timeoutCallBack(String clientId) {
        return () -> {
            log.info("連接超時：{}", clientId);
            removeUser(clientId);
        };
    }

    /**
     * 推送消息異常時，回調(diào)方法
     *
     * @param clientId 客戶端ID
     **/
    private Consumer<Throwable> errorCallBack(String clientId) {
        return throwable -> {
            log.error("SseEmitterServiceImpl[errorCallBack]：連接異常,客戶端ID:{}", clientId);

            // 推送消息失敗后，每隔10s推送一次，推送5次
            for (int i = 0; i < 5; i++) {
                try {
                    Thread.sleep(10000);
                    SseEmitter sseEmitter = sseCache.get(clientId);
                    if (sseEmitter == null) {
                        log.error("SseEmitterServiceImpl[errorCallBack]：第{}次消息重推失敗,未獲取到 {} 對應(yīng)的長鏈接", i + 1, clientId);
                        continue;
                    }
                    sseEmitter.send("失敗后重新推送");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        };
    }

    /**
     * 移除用戶連接
     *
     * @param clientId 客戶端ID
     **/
    private void removeUser(String clientId) {
        sseCache.remove(clientId);
        log.info("SseEmitterServiceImpl[removeUser]:移除用戶：{}", clientId);
    }
}

數(shù)據(jù)推送給前端

在 onEvent 回調(diào)中添加代碼，每接收到消息后就推送到前端

       
// 定義see接口
Request request = new Request.Builder().url("https://api.openai.com/v1/chat/completions")
        .header("Authorization","xxx")
        .post(okhttp3.RequestBody.create(okhttp3.MediaType.parse("application/json; charset=utf-8"),param.toJSONString()))
        .build();
OkHttpClient okHttpClient = new OkHttpClient.Builder()
        .connectTimeout(10, TimeUnit.MINUTES)
        .readTimeout(10, TimeUnit.MINUTES)//這邊需要將超時顯示設(shè)置長一點(diǎn)，不然剛連上就斷開，之前以為調(diào)用方式錯誤被坑了半天
        .build();

// 實(shí)例化EventSource，注冊EventSource監(jiān)聽器
RealEventSource realEventSource = new RealEventSource(request, new EventSourceListener() {

    @Override
    public void onOpen(EventSource eventSource, Response response) {
        log.info("onOpen");
    }

    @Override
    public void onEvent(EventSource eventSource, String id, String type, String data) {
		if ("[DONE]".equals(data)) {
            System.out.println("收到 [DONE] 信號");
            return;
        }
        ChatCompletionResp chatCompletionResp = JSON.parseObject(data, ChatCompletionResp.class);
        // 獲得生成的文章內(nèi)容
        if (CollUtil.isEmpty(chatCompletionResp.getChoices())){
            return;
        }
        Message delta = chatCompletionResp.getChoices().get(0).getDelta();
        if (delta == null){
            return;
        }

        sseEmitterService.sendMessageToOneClient(clientId , delta);
		log.info(data);//請求到的數(shù)據(jù)
    
    }

    @Override
    public void onClosed(EventSource eventSource) {
        log.info("onClosed");
//                emitter.complete();
    }

    @Override
    public void onFailure(EventSource eventSource, Throwable t, Response response) {
      log.error("onFailure 出現(xiàn)異常，response={}", response, t);//這邊可以監(jiān)聽并重新打開
//                emitter.complete();
    }
});
realEventSource.connect(okHttpClient);//真正開始請求的一步

前端代碼

由于 EventSource 不允許直接配置請求頭，普通的 EventSource 如果需要攜帶token請求,那就需要引入一個插件

安裝 EventSourcePolyfill

你可以通過npm安裝 event-source-polyfill:

npm install event-source-polyfill

引入 EventSourcePolyfill 后，它會自動替換瀏覽器中的原生 EventSource，其用法與原生的 API 一致。你可以像使用 EventSource 一樣使用它：

創(chuàng)建sseClient.js

封裝一下， sse 最佳實(shí)踐，

// utils/sseClient.js
import { EventSourcePolyfill } from 'event-source-polyfill'
import { baseURL } from '../config';


// 封裝一個創(chuàng)建 SSE 連接的方法
export function newEventSource({ clientId = '', headers = {}, onMessage, onError, onOpen }) {
    const token = sessionStorage.getItem('token') || ''

    const es = new EventSourcePolyfill(baseURL + 'p/sse/createConnect?clientId=' + clientId  , {
        headers: {
            'Authorization': `Bearer ${token}`
            ...headers
        },
        heartbeatTimeout: 60 * 1000, // 心跳超時（可選）
    })

    es.onopen = (event) => {
        console.log('SSE 連接已開啟')
        onOpen && onOpen(event)
    }

    es.onmessage = (event) => {
    	//前端：在接收到結(jié)束標(biāo)識后立即銷毀
        if (event.data === '[DONE]') {
            console.log('SSE 連接已關(guān)閉')
            es.close()
        }
        onMessage && onMessage(event)
    }

    es.onerror = (event) => {
        console.error('SSE 錯誤:', event)
        onError && onError(event)
        es.close() // 出錯時自動關(guān)閉
    }

    return es // 返回實(shí)例，方便外部主動關(guān)閉
}

vue3代碼

import { newEventSource } from '@/utils/sseClient.js'

const createSseConnection = () => {
  return newEventSource({
    clientId: 'xxx',
    onMessage: (event) => {
      console.log('Received SSE message:', event.data);
    }
  });
};

優(yōu)化后端代碼

按需建立連接并及時關(guān)閉是非常關(guān)鍵的實(shí)踐策略，每一個 SseEmitter 在服務(wù)端都是一個線程或者任務(wù)掛起的狀態(tài)，太多不關(guān)閉會導(dǎo)致資源消耗（線程、連接、內(nèi)存等）；

如果每個用戶長時間掛一個 SSE，不及時關(guān)閉，可能造成內(nèi)存泄露或線程池耗盡，所以我們優(yōu)化一下后端代碼，在完成輸出后及時關(guān)閉連接.

在關(guān)閉和異常的回調(diào)方法中添加：

sseEmitterService.sendMessageToOneClient(clientId, "[DONE]");
sseEmitterService.closeConnect(clientId);

修改后：

       
// 定義see接口
Request request = new Request.Builder().url("https://api.openai.com/v1/chat/completions")
        .header("Authorization","xxx")
        .post(okhttp3.RequestBody.create(okhttp3.MediaType.parse("application/json; charset=utf-8"),param.toJSONString()))
        .build();
OkHttpClient okHttpClient = new OkHttpClient.Builder()
        .connectTimeout(10, TimeUnit.MINUTES)
        .readTimeout(10, TimeUnit.MINUTES)//這邊需要將超時顯示設(shè)置長一點(diǎn)，不然剛連上就斷開，之前以為調(diào)用方式錯誤被坑了半天
        .build();

// 實(shí)例化EventSource，注冊EventSource監(jiān)聽器
RealEventSource realEventSource = new RealEventSource(request, new EventSourceListener() {

    @Override
    public void onOpen(EventSource eventSource, Response response) {
        log.info("onOpen");
    }

    @SneakyThrows
    @Override
    public void onEvent(EventSource eventSource, String id, String type, String data) {
		// log.info("onEvent");
		// 在實(shí)際應(yīng)用中，你可以在這里將數(shù)據(jù)推送給前端
        log.info(data);//請求到的數(shù)據(jù)
    }

    @Override
    public void onClosed(EventSource eventSource) {
        log.info("onClosed");
        sseEmitterService.sendMessageToOneClient(clientId, "[DONE]");
        sseEmitterService.closeConnect(clientId);
//                emitter.complete();
    }

    @Override
    public void onFailure(EventSource eventSource, Throwable t, Response response) {
        log.error("onFailure 出現(xiàn)異常，response={}", response, t);//這邊可以監(jiān)聽并重新打開
        sseEmitterService.sendMessageToOneClient(clientId, "[DONE]");
        sseEmitterService.closeConnect(clientId);
//                emitter.complete();
    }
});
realEventSource.connect(okHttpClient);//真正開始請求的一步

輸出效果如下：

踩坑

sseEmitterService.sendMessageToOneClient(clientId, "[DONE]");

使用該方法發(fā)送純文本時，里面不能有任何回車符 \n，因?yàn)榘l(fā)送到前端后換行符可能被當(dāng)作字段結(jié)束符處理，最好封裝對象然后轉(zhuǎn)JSON字符串發(fā)送

總結(jié)

以上為個人經(jīng)驗(yàn)，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章:

最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线