优化:添加阻塞队列,优化缓存禁止

This commit is contained in:
JianYu Zheng
2025-08-14 14:44:52 +08:00
parent a9d116df2f
commit 7c0df908e9
2 changed files with 116 additions and 39 deletions
@@ -0,0 +1,91 @@
export default class BlockingQueue {
#items = [];
#waiters = []; // {resolve, reject, min, timer, onTimeout}
/* 空队列一次性闸门 */
#emptyPromise = null;
#emptyResolve = null;
/* 生产者:把数据塞进去 */
enqueue(item) {
this.#items.push(item);
// 若有空队列闸门,一次性放行所有等待者
if (this.#emptyResolve) {
this.#emptyResolve();
this.#emptyResolve = null;
this.#emptyPromise = null;
}
// 唤醒所有正在等的 waiter
this.#wakeWaiters();
}
/* 消费者:min 条或 timeout ms 先到谁 */
async dequeue(min = 1, timeout = Infinity, onTimeout = null) {
// 1. 若空,等第一次数据到达(所有调用共享同一个 promise)
if (this.#items.length === 0) {
await this.#waitForFirstItem();
}
// 立即满足
if (this.#items.length >= min) {
return this.#flush();
}
// 需要等待
return new Promise((resolve, reject) => {
let timer = null;
const waiter = { resolve, reject, min, onTimeout, timer };
// 超时逻辑
if (Number.isFinite(timeout)) {
waiter.timer = setTimeout(() => {
this.#removeWaiter(waiter);
if (onTimeout) onTimeout(this.#items.length);
resolve(this.#flush());
}, timeout);
}
this.#waiters.push(waiter);
});
}
/* 空队列闸门生成器 */
#waitForFirstItem() {
if (!this.#emptyPromise) {
this.#emptyPromise = new Promise(r => (this.#emptyResolve = r));
}
return this.#emptyPromise;
}
/* 内部:每次数据变动后,检查哪些 waiter 已满足 */
#wakeWaiters() {
for (let i = this.#waiters.length - 1; i >= 0; i--) {
const w = this.#waiters[i];
if (this.#items.length >= w.min) {
this.#removeWaiter(w);
w.resolve(this.#flush());
}
}
}
#removeWaiter(waiter) {
const idx = this.#waiters.indexOf(waiter);
if (idx !== -1) {
this.#waiters.splice(idx, 1);
if (waiter.timer) clearTimeout(waiter.timer);
}
}
#flush() {
const snapshot = [...this.#items];
this.#items.length = 0;
return snapshot;
}
/* 当前缓存长度(不含等待者) */
get length() {
return this.#items.length;
}
}
+25 -39
View File
@@ -104,6 +104,7 @@
import { webSocketConnect } from './js/xiaoZhiConnect.js'; import { webSocketConnect } from './js/xiaoZhiConnect.js';
import { checkOpusLoaded , initOpusEncoder } from './js/opus.js'; import { checkOpusLoaded , initOpusEncoder } from './js/opus.js';
import { addMessage } from './js/document.js' import { addMessage } from './js/document.js'
import BlockingQueue from './js/utils/BlockingQueue.js'
// 需要加载的脚本列表 - 移除Opus依赖 // 需要加载的脚本列表 - 移除Opus依赖
const scriptFiles = []; const scriptFiles = [];
@@ -134,7 +135,6 @@
let totalAudioSize = 0; // 跟踪累积的音频大小 let totalAudioSize = 0; // 跟踪累积的音频大小
let audioBufferQueue = []; // 存储接收到的音频包 let audioBufferQueue = []; // 存储接收到的音频包
let isAudioBuffering = false; // 是否正在缓冲音频
let isAudioPlaying = false; // 是否正在播放音频 let isAudioPlaying = false; // 是否正在播放音频
const BUFFER_THRESHOLD = 3; // 缓冲包数量阈值,至少累积3个包再开始播放 const BUFFER_THRESHOLD = 3; // 缓冲包数量阈值,至少累积3个包再开始播放
const MIN_AUDIO_DURATION = 0.1; // 最小音频长度(秒),小于这个长度的音频会被合并 const MIN_AUDIO_DURATION = 0.1; // 最小音频长度(秒),小于这个长度的音频会被合并
@@ -188,13 +188,9 @@
} }
} }
const queue = new BlockingQueue();
// 开始音频缓冲过程 // 开始音频缓冲过程
function startAudioBuffering() { async function startAudioBuffering() {
if (isAudioBuffering || isAudioPlaying) return;
isAudioBuffering = true;
log("开始音频缓冲...", 'info'); log("开始音频缓冲...", 'info');
// 先尝试初始化解码器,以便在播放时已准备好 // 先尝试初始化解码器,以便在播放时已准备好
@@ -202,29 +198,29 @@
log(`预初始化Opus解码器失败: ${error.message}`, 'warning'); log(`预初始化Opus解码器失败: ${error.message}`, 'warning');
// 继续缓冲,我们会在播放时再次尝试初始化 // 继续缓冲,我们会在播放时再次尝试初始化
}); });
const timeout = 300;
// 设置超时,如果在一定时间内没有收集到足够的音频包,就开始播放 while (true) {
setTimeout(() => { const packets = await queue.dequeue(
if (isAudioBuffering && audioBufferQueue.length > 0) { 3, // 至少 3 条
log(`缓冲超时,当前缓冲包数: ${audioBufferQueue.length},开始播放`, 'info'); timeout, // 最多等 300 ms
(count) => { // 超时额外回调
log(`缓冲超时,当前缓冲包数: ${count},开始播放`, 'info');
}
);
if (packets.length) {
log(`已缓冲 ${packets.length} 个音频包,开始播放`, 'info');
audioBufferQueue.push(...packets)
playBufferedAudio(); playBufferedAudio();
} }
}, 300); // 300ms超时 console.log(queue.length)
while (true){
// 监控缓冲进度 const data = await queue.dequeue(99,50)
const bufferCheckInterval = setInterval(() => { if(data.length){
if (!isAudioBuffering) { audioBufferQueue.push(...data)
clearInterval(bufferCheckInterval); playBufferedAudio();
return; }
} }
}
// 当累积了足够的音频包,开始播放
if (audioBufferQueue.length >= BUFFER_THRESHOLD) {
clearInterval(bufferCheckInterval);
log(`已缓冲 ${audioBufferQueue.length} 个音频包,开始播放`, 'info');
playBufferedAudio();
}
}, 50);
} }
// 播放已缓冲的音频 // 播放已缓冲的音频
@@ -232,7 +228,6 @@
if (isAudioPlaying || audioBufferQueue.length === 0) return; if (isAudioPlaying || audioBufferQueue.length === 0) return;
isAudioPlaying = true; isAudioPlaying = true;
isAudioBuffering = false;
// 确保Opus解码器已初始化 // 确保Opus解码器已初始化
const initDecoderAndPlay = async () => { const initDecoderAndPlay = async () => {
@@ -360,7 +355,6 @@
// 使用setTimeout避免递归调用 // 使用setTimeout避免递归调用
setTimeout(() => { setTimeout(() => {
console.log(this.i++,this.queue.length ,audioBufferQueue.length)
// 如果队列中还有数据,继续播放 // 如果队列中还有数据,继续播放
if (this.queue.length > 0) { if (this.queue.length > 0) {
this.startPlaying(); this.startPlaying();
@@ -378,7 +372,6 @@
} else { } else {
// 等待更多数据 // 等待更多数据
setTimeout(() => { setTimeout(() => {
console.log(this.i,this.queue.length ,audioBufferQueue.length)
// 如果仍然没有新数据,但有更多的包到达 // 如果仍然没有新数据,但有更多的包到达
if (this.queue.length === 0 && audioBufferQueue.length > 0) { if (this.queue.length === 0 && audioBufferQueue.length > 0) {
const frames = [...audioBufferQueue]; const frames = [...audioBufferQueue];
@@ -999,7 +992,6 @@
if (message === '' || !websocket || websocket.readyState !== WebSocket.OPEN) return; if (message === '' || !websocket || websocket.readyState !== WebSocket.OPEN) return;
audioBufferQueue = []; audioBufferQueue = [];
isAudioBuffering = false;
isAudioPlaying = false; isAudioPlaying = false;
try { try {
@@ -1252,6 +1244,7 @@
}).catch(error => { }).catch(error => {
log(`Opus解码器预加载失败: ${error.message},将在需要时重试`, 'warning'); log(`Opus解码器预加载失败: ${error.message},将在需要时重试`, 'warning');
}); });
startAudioBuffering()
} }
// PCM录音处理器代码 - 会被注入到AudioWorklet中 // PCM录音处理器代码 - 会被注入到AudioWorklet中
@@ -1680,18 +1673,11 @@
log(`收到未知类型的二进制数据: ${typeof data}`, 'warning'); log(`收到未知类型的二进制数据: ${typeof data}`, 'warning');
return; return;
} }
// 创建Uint8Array用于处理 // 创建Uint8Array用于处理
const opusData = new Uint8Array(arrayBuffer); const opusData = new Uint8Array(arrayBuffer);
console.log(audioBufferQueue.length)
if (opusData.length > 0) { if (opusData.length > 0) {
// 将数据添加到缓冲队列 // 将数据添加到缓冲队列
audioBufferQueue.push(opusData); queue.enqueue(opusData);
// 如果收到的是第一个音频包,开始缓冲过程
if (audioBufferQueue.length === 1 && !isAudioBuffering && !isAudioPlaying) {
startAudioBuffering();
}
} else { } else {
log('收到空音频数据帧,可能是结束标志', 'warning'); log('收到空音频数据帧,可能是结束标志', 'warning');