Kivy安卓端如何通过Pyjnius实现麦克风/扬声器音频流处理

实现逻辑说明 #

安卓系统原生提供AudioRecord（麦克风采集）、AudioTrack（PCM音频播放）两个底层音频API，通过Pyjnius可以直接调用这两个类实现和PyAudio完全一致的字节流读写能力，不需要依赖维护状态不佳的Plyer、audiostream等第三方库，跨端业务逻辑复用率可达90%以上。

前置配置 #

首先在buildozer.spec文件中补充安卓必要权限：

• 麦克风采集权限：android.permission.RECORD_AUDIO
• 防止采集时息屏被系统杀进程：android.permission.WAKE_LOCK
• 安卓6.0及以上版本需要在应用启动时主动申请动态权限，不能依赖系统自动弹窗授权。

具体代码实现 #

首先导入依赖模块：

from jnius import autoclass
from android.permissions import request_permissions, Permission
import threading
import time
import sys

# 加载安卓原生音频相关Java类
AudioFormat = autoclass('android.media.AudioFormat')
AudioRecord = autoclass('android.media.AudioRecord')
AudioTrack = autoclass('android.media.AudioTrack')
AudioManager = autoclass('android.media.AudioManager')
ByteBuffer = autoclass('java.nio.ByteBuffer')

麦克风采集类（对齐PyAudio读流逻辑） #

注意不要使用Java层回调接口取数据：这类回调默认投递到主线程，会被Kivy的GL渲染事件循环阻塞，这也是audiostream示例回调始终不触发的核心原因。直接在独立工作线程中同步读字节，稳定性和PyAudio完全一致。

class AndroidAudioRecorder:
    def __init__(self, sample_rate=16000, channels=1, bit_depth=16, chunk_size=1024):
        self.sample_rate = sample_rate
        self.chunk_size = chunk_size
        self.frame_byte_length = chunk_size * (bit_depth // 8)
        self.is_running = False
        self.worker_thread = None

        # 匹配安卓原生音频参数
        channel_config = AudioFormat.CHANNEL_IN_MONO if channels == 1 else AudioFormat.CHANNEL_IN_STEREO
        audio_format = AudioFormat.ENCODING_PCM_16BIT if bit_depth == 16 else AudioFormat.ENCODING_PCM_8BIT
        min_buffer = AudioRecord.getMinBufferSize(sample_rate, channel_config, audio_format)
        # 实际缓冲区取系统最小值的2倍，避免丢帧
        self.buffer_size = max(min_buffer, self.frame_byte_length * 2)

        self.recorder = AudioRecord(
            AudioManager.STREAM_MUSIC,
            sample_rate,
            channel_config,
            audio_format,
            self.buffer_size
        )

    def _read_loop(self, data_callback):
        self.recorder.startRecording()
        temp_buffer = ByteBuffer.allocate(self.frame_byte_length)
        while self.is_running:
            read_len = self.recorder.read(temp_buffer, self.frame_byte_length)
            if read_len > 0:
                # 转为Python原生bytes对象，直接传入业务逻辑
                pcm_data = bytes(temp_buffer.array())[:read_len]
                data_callback(pcm_data)
            temp_buffer.clear()
        self.recorder.stop()
        self.recorder.release()

    def start(self, data_callback):
        # 申请动态麦克风权限
        request_permissions([Permission.RECORD_AUDIO])
        # 等待录音器初始化完成
        while self.recorder.getState() != AudioRecord.STATE_INITIALIZED:
            time.sleep(0.1)
        self.is_running = True
        self.worker_thread = threading.Thread(target=self._read_loop, args=(data_callback,), daemon=True)
        self.worker_thread.start()

    def stop(self):
        self.is_running = False
        if self.worker_thread:
            self.worker_thread.join()

扬声器播放类（对齐PyAudio写流逻辑） #

同样采用独立线程同步写的模式，避免主线程阻塞：

class AndroidAudioPlayer:
    def __init__(self, sample_rate=16000, channels=1, bit_depth=16, chunk_size=1024):
        self.sample_rate = sample_rate
        self.chunk_size = chunk_size
        self.is_running = False
        self.worker_thread = None
        self.pcm_queue = []

        channel_config = AudioFormat.CHANNEL_OUT_MONO if channels == 1 else AudioFormat.CHANNEL_OUT_STEREO
        audio_format = AudioFormat.ENCODING_PCM_16BIT if bit_depth == 16 else AudioFormat.ENCODING_PCM_8BIT
        min_buffer = AudioTrack.getMinBufferSize(sample_rate, channel_config, audio_format)
        self.buffer_size = max(min_buffer, chunk_size * (bit_depth//8) * 2)

        self.player = AudioTrack(
            AudioManager.STREAM_MUSIC,
            sample_rate,
            channel_config,
            audio_format,
            self.buffer_size,
            AudioTrack.MODE_STREAM
        )

    def _write_loop(self):
        self.player.play()
        while self.is_running or len(self.pcm_queue) > 0:
            if not self.pcm_queue:
                time.sleep(0.001)
                continue
            pcm_data = self.pcm_queue.pop(0)
            write_buf = ByteBuffer.wrap(pcm_data)
            # 阻塞式写入，保证播放流畅不爆音
            self.player.write(write_buf, len(pcm_data), AudioTrack.WRITE_BLOCKING)
        self.player.stop()
        self.player.release()

    def start(self):
        self.is_running = True
        self.worker_thread = threading.Thread(target=self._write_loop, daemon=True)
        self.worker_thread.start()

    def write(self, pcm_data):
        # 对外暴露和PyAudio完全一致的写接口
        self.pcm_queue.append(pcm_data)

    def stop(self):
        self.is_running = False
        if self.worker_thread:
            self.worker_thread.join()

跨端兼容封装 #

直接在入口处判断运行平台，对外暴露统一的音频类接口，上层业务逻辑不需要做平台适配：

if sys.platform in ("win32", "linux"):
    # 替换为你原有PyAudio的封装实现
    from pyaudio_backend import PyAudioRecorder as AudioRecorder, PyAudioPlayer as AudioPlayer
elif sys.platform == "android":
    # 直接使用上述安卓实现
    AudioRecorder = AndroidAudioRecorder
    AudioPlayer = AndroidAudioPlayer

注意事项 #

• 缓冲区大小不要自定义硬编码，必须先调用getMinBufferSize获取当前设备支持的最小缓冲区，否则会出现AudioRecord/AudioTrack初始化失败、无声等问题。
• 所有音频读写操作必须放在独立守护线程中运行，绝对不能在Kivy主线程执行读写，否则会直接卡住UI。
• 如果需要适配安卓12+的蓝牙麦克风采集，需要额外补充android.permission.BLUETOOTH_CONNECT权限。

内容的提问来源于stack exchange，提问作者user name