实时音频流中socket.io引发的内存泄漏问题及优化策略

import pyaudio
import numpy as np
import socketio
import threading
import time

# 假设sio已经被正确初始化
sio = socketio.Client()
sio.connect('http://localhost:5000') # 替换为你的服务器地址

class AudioStreamer:
    def __init__(self):
        self.CHANNELS = 1
        self.CHUNK = 1024 # 音频帧大小
        self.is_running = True
        self.audio_buffer = [] # 用于缓存音频数据
        self.BUFFER_SIZE = 10 # 缓存10个CHUNK的数据后发送
        self.BUFFER_TIME_SECONDS = 0.5 # 或者每0.5秒发送一次

    def send_audio_e(self):
        p = pyaudio.PyAudio()
        stream = p.open(
            format=pyaudio.paInt16,
            channels=self.CHANNELS,
            rate=44100,
            input=True,
            frames_per_buffer=self.CHUNK,
        )

        last_send_time = time.time()

        try:
            while self.is_running:
                data = stream.read(self.CHUNK, exception_on_overflow=False) # 避免溢出报错
                audio_data_chunk = np.frombuffer(data, dtype=np.int16).tobytes()

                self.audio_buffer.append(audio_data_chunk)

                # 达到缓冲区大小或达到发送时间间隔，则发送
                if len(self.audio_buffer) >= self.BUFFER_SIZE or \
                   (time.time() - last_send_time) >= self.BUFFER_TIME_SECONDS:

                    # 将所有缓存的音频块合并成一个更大的数据块发送
                    # 注意：这里简单地拼接，接收端需要知道如何解析
                    # 更复杂的场景可能需要自定义协议头
                    combined_audio_data = b"".join(self.audio_buffer)

                    try:
                        sio.emit("audio_data", {"audio_data": combined_audio_data})
                        # 显式清除内存，帮助垃圾回收
                        combined_audio_data = None 
                    except Exception as e:
                        print(f"Error emitting audio data: {e}")

                    self.audio_buffer.clear() # 清空缓冲区
                    last_send_time = time.time() # 更新发送时间

                # 避免CPU空转过高，可以适当添加短暂停顿
                # time.sleep(0.001) 
        except Exception as e:
            print(f"Error in send_audio_e: {e}")
        finally:
            print("Audio stream CLOSED")
            stream.stop_stream()
            stream.close()
            p.terminate()
            sio.disconnect() # 在线程结束时断开连接

    def start_communication(self):
        threading.Thread(target=self.send_audio_e).start()

    def stop_communication(self):
        self.is_running = False

# 示例用法
if __name__ == "__main__":
    streamer = AudioStreamer()
    streamer.start_communication()
    try:
        # 让主线程运行一段时间，模拟应用生命周期
        time.sleep(60) 
    except KeyboardInterrupt:
        print("Stopping streamer...")
    finally:
        streamer.stop_communication()
        print("Streamer stopped.")

# 在上述示例代码中已包含此优化
# ...
                    try:
                        sio.emit("audio_data", {"audio_data": combined_audio_data})
                        # 显式清除内存，帮助垃圾回收
                        combined_audio_data = None 
                    except Exception as e:
                        print(f"Error emitting audio data: {e}")
# ...