如何将字节数组数据流接入androidx.media3.exoplayer？

安卓应用需要播放从其他应用接收的字节数组格式视频流，使用androidx.media3.exoplayer实现播放。需求是仅初始化一次ExoPlayer实例，后续每次收到字节数组数据就传入播放器，但现有实现运行崩溃，求正确的自定义数据源实现方案。

尝试的代码 #

@OptIn(UnstableApi::class)
private suspend fun prepareExoPlayerFromByteArray(data: ByteArray) {

    if(data.isNotEmpty()) {
        Log.d(TAG, ">> Data is not empty $data")
        val extractorFactory = ExtractorsFactory { arrayOf(Ac3Extractor()) }
        val dsf = CustomDataSourceFactory(data)
        dsf.createDataSource()
        val mediasourceFactory = ProgressiveMediaSource.Factory(dsf,
            extractorFactory
        )
        withContext(Dispatchers.Main) {
            exoPlayer = ExoPlayer.Builder(this@MainActivity, mediasourceFactory).build()
            exoPlayer!!.setMediaItem(MediaItem.fromUri(Uri.EMPTY))
            // Prepare the player.
            exoPlayer!!.prepare()
            // Start the playback.
            exoPlayer!!.play()
        }

    } else {
        Log.d(TAG, ">> Data is empty $data")
    }
}

崩溃报错信息 #

Playback error
androidx.media3.exoplayer.ExoPlaybackException: Source error
androidx.media3.exoplayer.ExoPlayerImplInternal.handleIoException(ExoPlayerImplInternal.java:701)
androidx.media3.exoplayer.ExoPlayerImplInternal.handleMessage(ExoPlayerImplInternal.java:671)
at android.os.Handler.dispatchMessage(Handler.java:102)
at android.os.Looper.loopOnce(Looper.java:201)
at android.os.Looper.loop(Looper.java:288)
at android.os.HandlerThread.run(HandlerThread.java:67)
Caused by: androidx.media3.common.ParserException: Loading finished before preparation is complete.{contentIsMalformed=true, dataType=1}
at androidx.media3.exoplayer.source.ProgressiveMediaPeriod.maybeThrowPrepareError(ProgressiveMediaPeriod.java:247)
at androidx.media3.exoplayer.source.MaskingMediaPeriod.maybeThrowPrepareError(MaskingMediaPeriod.java:157)
at androidx.media3.exoplayer.ExoPlayerImplInternal.doSomeWork(ExoPlayerImplInternal.java:1096)

解决方案 #

1. 实现支持字节数组流的自定义DataSource #

继承BaseDataSource，实现线程安全的动态写入/读取逻辑：

@OptIn(UnstableApi::class)
class ByteArrayDataSource : BaseDataSource(true) {
    private val byteBuffer = ByteBuffer.allocate(1024 * 1024) // 1MB缓冲可按需调整
    private var isOpen = false
    private var readPosition = 0L

    @Synchronized
    fun writeData(data: ByteArray) {
        byteBuffer.put(data)
        byteBuffer.flip()
        notifyAll() // 通知读取线程有新数据
    }

    override fun open(dataSpec: DataSpec): Long {
        transferInitializing(dataSpec)
        isOpen = true
        transferStarted(dataSpec)
        return C.LENGTH_UNSET // 流式数据长度未知
    }

    override fun read(buffer: ByteArray, offset: Int, length: Int): Int {
        check(isOpen) { "DataSource not open" }
        synchronized(this) {
            while (byteBuffer.remaining() == 0) {
                try {
                    wait() // 无数据时等待
                } catch (e: InterruptedException) {
                    Thread.currentThread().interrupt()
                    return C.RESULT_END_OF_INPUT
                }
            }
            val readBytes = min(length, byteBuffer.remaining())
            byteBuffer.get(buffer, offset, readBytes)
            readPosition += readBytes
            bytesTransferred(readBytes)
            return readBytes
        }
    }

    override fun getUri(): Uri? = null

    override fun close() {
        isOpen = false
        synchronized(this) {
            notifyAll()
            byteBuffer.clear()
        }
    }
}

2. 实现对应的DataSourceFactory #

class ByteArrayDataSourceFactory : DataSource.Factory {
    private val dataSource = ByteArrayDataSource()

    fun getDataSource(): ByteArrayDataSource = dataSource

    override fun createDataSource(): DataSource = dataSource
}

3. 正确初始化ExoPlayer并动态传入数据 #

仅初始化一次ExoPlayer，后续收到字节数组时直接写入数据源：

private lateinit var exoPlayer: ExoPlayer
private lateinit var byteArrayDataSourceFactory: ByteArrayDataSourceFactory

// 在Activity/Fragment初始化时调用（如onCreate）
private fun initExoPlayer() {
    byteArrayDataSourceFactory = ByteArrayDataSourceFactory()
    val extractorsFactory = ExtractorsFactory {
        arrayOf(
            DefaultExtractor(), // 自动适配多数格式，也可指定具体Extractor如Ac3Extractor
            Ac3Extractor()
        )
    }
    val mediaSourceFactory = ProgressiveMediaSource.Factory(
        byteArrayDataSourceFactory,
        extractorsFactory
    )
    exoPlayer = ExoPlayer.Builder(this)
        .setMediaSourceFactory(mediaSourceFactory)
        .build()
    // 绑定PlayerView（如果有）
    playerView.player = exoPlayer
}

// 每次收到字节数组时调用
private fun onReceiveVideoData(data: ByteArray) {
    if (data.isEmpty()) return
    // 写入数据到自定义数据源
    byteArrayDataSourceFactory.getDataSource().writeData(data)
    // 如果播放器未准备，触发准备
    if (!exoPlayer.isPlaying && !exoPlayer.isLoading) {
        val mediaItem = MediaItem.fromUri(Uri.EMPTY) // 自定义数据源不需要真实Uri
        exoPlayer.setMediaItem(mediaItem)
        exoPlayer.prepare()
        exoPlayer.play()
    }
}

关键注意点 #

• 自定义ByteArrayDataSource必须处理线程同步，避免写入/读取冲突
• 使用C.LENGTH_UNSET告知ExoPlayer这是无固定长度的流式数据
• 仅初始化一次ExoPlayer，避免重复创建实例导致的资源浪费
• 通用场景用DefaultExtractor自动适配格式，固定格式可指定具体Extractor
• 首次收到有效数据后再触发播放器的prepare/play，避免空数据源解析错误

内容的提问来源于stack exchange，提问作者Nidheesh Kumar S V