局域網TCP通過組播放地址rtp推流和拉流實現實時喊話

應用場景，安卓端局域網不用ip通過組播放地址實現實時對講功能

發送端: ffmpeg -f alsa -i hw:1 -acodec aac -ab 64k -ac 2 -ar 16000 -frtp -sdp file stream.sdp rtp://224.0.0.1:14556

接收端: ffmpeg -protocol whitelist file,udp,rtp -i stream.sdp -acodec pcm s16le -ar 16000 -ac 2 -f alsa default

在windows上測試通過后然后在安卓中實現

# 查詢本地可用麥克風設備
ffmpeg -list_devices true -f dshow -i dummy

麥克風 (Realtek(R) Audio)這是我電腦的
# windows? 執行RTM推音頻流
ffmpeg -f dshow -i audio="麥克風 (Realtek(R) Audio)" -acodec aac -ab 64k -ac 2 -ar 16000 -f rtp -sdp_file stream.sdp rtp://239.0.0.1:15556

上面windows上調通后接下來在安卓上實現

implementation("com.arthenica:mobile-ffmpeg-full:4.4.LTS")主要用到這個庫

package com.xhx.megaphone.tcpimport android.media.AudioFormat
import android.media.AudioRecord
import android.media.MediaRecorder
import com.arthenica.mobileffmpeg.Config
import com.arthenica.mobileffmpeg.FFmpeg
import com.blankj.utilcode.util.LogUtils
import com.xhx.megaphone.App
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch
import java.io.File
import java.io.FileOutputStream
import java.io.IOException
import java.util.concurrent.atomic.AtomicBoolean/*** 實時推流助手 - 最優化版本* * 功能：* 1. 錄音buffer直接實時寫入臨時文件* 2. FFmpeg同時讀取文件進行推流* 3. 最小化延遲的實時推流*/
object LiveStreamingHelper {private const val TAG = "LiveStreamingHelper"// 組播配置private const val MULTICAST_ADDRESS = "239.0.0.1"private const val MULTICAST_PORT = 15556// 音頻參數private const val SAMPLE_RATE = 16000private const val CHANNELS = 1private const val BIT_RATE = 64000private const val AUDIO_FORMAT = AudioFormat.ENCODING_PCM_16BIT// 緩沖區大小 - 使用較小的緩沖區減少延遲private val BUFFER_SIZE = AudioRecord.getMinBufferSize(SAMPLE_RATE,AudioFormat.CHANNEL_IN_MONO,AUDIO_FORMAT).let { minSize ->// 使用最小緩沖區的2倍，減少延遲minSize * 2}// 推流狀態private val isStreaming = AtomicBoolean(false)private var audioRecord: AudioRecord? = nullprivate var recordingThread: Thread? = nullprivate var ffmpegExecutionId: Long = 0// 文件路徑private val cacheDir = File(App.ctx.cacheDir, "live_streaming")private val sdpFile = File(cacheDir, "stream.sdp")private val liveAudioFile = File(cacheDir, "live_audio.pcm")/*** 開始實時錄音推流*/fun startStreaming(): Boolean {if (isStreaming.get()) {LogUtils.w(TAG, "推流已在進行中")return false}return try {initializeFiles()createSdpFile()startAudioRecording()startLiveStreaming()isStreaming.set(true)LogUtils.i(TAG, "? 實時推流啟動成功")true} catch (e: Exception) {LogUtils.e(TAG, "? 實時推流啟動失敗", e)stopStreaming()false}}/*** 停止推流*/fun stopStreaming() {if (!isStreaming.get()) {return}isStreaming.set(false)// 停止錄音audioRecord?.stop()audioRecord?.release()audioRecord = null// 停止錄音線程recordingThread?.interrupt()recordingThread = null// 停止FFmpegif (ffmpegExecutionId != 0L) {FFmpeg.cancel(ffmpegExecutionId)ffmpegExecutionId = 0}LogUtils.i(TAG, "🛑 實時推流已停止")}/*** 獲取推流狀態*/fun isStreaming(): Boolean = isStreaming.get()/*** 獲取SDP文件路徑*/fun getSdpFilePath(): String = sdpFile.absolutePath/*** 獲取組播地址信息*/fun getMulticastInfo(): String {val fileSize = if (liveAudioFile.exists()) {"${liveAudioFile.length() / 1024}KB"} else {"0KB"}return "組播地址: $MULTICAST_ADDRESS:$MULTICAST_PORT\n" +"SDP文件: ${sdpFile.absolutePath}\n" +"傳輸方式: 實時文件流\n" +"緩沖區大小: ${BUFFER_SIZE}字節\n" +"當前文件大小: $fileSize\n" +"推流狀態: ${if (isStreaming.get()) "進行中" else "已停止"}"}/*** 初始化文件和目錄*/private fun initializeFiles() {if (!cacheDir.exists()) {cacheDir.mkdirs()}// 清理舊文件if (liveAudioFile.exists()) {liveAudioFile.delete()}// 創建新的音頻文件liveAudioFile.createNewFile()}/*** 創建SDP文件*/private fun createSdpFile() {val sdpContent = """v=0o=- 0 0 IN IP4 127.0.0.1s=No Namec=IN IP4 $MULTICAST_ADDRESSt=0 0a=tool:libavformat 58.45.100m=audio $MULTICAST_PORT RTP/AVP 97b=AS:64a=rtpmap:97 MPEG4-GENERIC/$SAMPLE_RATE/$CHANNELSa=fmtp:97 profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3; config=141056E500""".trimIndent()sdpFile.writeText(sdpContent)LogUtils.i(TAG, "SDP文件創建成功: ${sdpFile.absolutePath}")}/*** 開始音頻錄音 - 直接實時寫入文件*/private fun startAudioRecording() {audioRecord = AudioRecord(MediaRecorder.AudioSource.MIC,SAMPLE_RATE,AudioFormat.CHANNEL_IN_MONO,AUDIO_FORMAT,BUFFER_SIZE)if (audioRecord?.state != AudioRecord.STATE_INITIALIZED) {throw IOException("AudioRecord初始化失敗")}audioRecord?.startRecording()// 啟動錄音線程，實時寫入文件recordingThread = Thread {val buffer = ByteArray(BUFFER_SIZE)var fileOutputStream: FileOutputStream? = nullvar totalBytes = 0var lastLogTime = System.currentTimeMillis()try {fileOutputStream = FileOutputStream(liveAudioFile, false) // 不追加，覆蓋寫入LogUtils.i(TAG, "錄音線程啟動，實時寫入: ${liveAudioFile.absolutePath}")LogUtils.i(TAG, "緩沖區大小: $BUFFER_SIZE 字節")while (isStreaming.get() && !Thread.currentThread().isInterrupted) {val bytesRead = audioRecord?.read(buffer, 0, buffer.size) ?: 0if (bytesRead > 0) {// 立即寫入文件并刷新fileOutputStream.write(buffer, 0, bytesRead)fileOutputStream.flush()totalBytes += bytesRead// 每3秒打印一次狀態（更頻繁的狀態更新）val currentTime = System.currentTimeMillis()if (currentTime - lastLogTime > 3000) {LogUtils.d(TAG, "🎙? 實時錄音中: ${totalBytes / 1024}KB, 速率: ${(totalBytes / ((currentTime - (lastLogTime - 3000)) / 1000.0) / 1024).toInt()}KB/s")lastLogTime = currentTime}} else if (bytesRead == AudioRecord.ERROR_INVALID_OPERATION) {LogUtils.e(TAG, "AudioRecord讀取錯誤: ERROR_INVALID_OPERATION")break} else if (bytesRead < 0) {LogUtils.w(TAG, "AudioRecord讀取返回負值: $bytesRead")}}LogUtils.i(TAG, "錄音線程結束，總計: ${totalBytes / 1024}KB")} catch (e: Exception) {LogUtils.e(TAG, "錄音數據寫入異常", e)} finally {fileOutputStream?.close()}}recordingThread?.start()LogUtils.i(TAG, "音頻錄音已啟動")}/*** 啟動實時推流*/private fun startLiveStreaming() {GlobalScope.launch(Dispatchers.IO) {// 等待一些音頻數據寫入Thread.sleep(300)// 構建FFmpeg命令 - 使用較小的緩沖區和實時參數val command = "-re -f s16le -ar $SAMPLE_RATE -ac $CHANNELS " +"-thread_queue_size 512 " +  // 增加線程隊列大小"-i ${liveAudioFile.absolutePath} " +"-acodec aac -ab ${BIT_RATE/1000}k -ac $CHANNELS -ar $SAMPLE_RATE " +"-f rtp -sdp_file ${sdpFile.absolutePath} " +"rtp://$MULTICAST_ADDRESS:$MULTICAST_PORT"LogUtils.i(TAG, "FFmpeg實時推流命令: $command")ffmpegExecutionId = FFmpeg.executeAsync(command) { executionId, returnCode ->LogUtils.i(TAG, "FFmpeg推流結束: executionId=$executionId, returnCode=$returnCode")when (returnCode) {Config.RETURN_CODE_SUCCESS -> {LogUtils.i(TAG, "? 推流正常結束")}Config.RETURN_CODE_CANCEL -> {LogUtils.i(TAG, "🛑 推流被用戶取消")}else -> {LogUtils.w(TAG, "?? 推流異常結束，返回碼: $returnCode")}}}LogUtils.i(TAG, "FFmpeg執行ID: $ffmpegExecutionId")}}
}

拉流

package com.xhx.megaphone.tcpimport android.media.AudioFormat
import android.media.AudioManager
import android.media.AudioTrack
import com.arthenica.mobileffmpeg.Config
import com.arthenica.mobileffmpeg.FFmpeg
import com.blankj.utilcode.util.LogUtils
import com.xhx.megaphone.App
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch
import java.io.File
import java.io.RandomAccessFile
import java.util.concurrent.atomic.AtomicBoolean
import java.util.concurrent.atomic.AtomicLong/*** 低延遲拉流播放助手* * 優化策略：* 1. 最小化FFmpeg緩沖* 2. 減少AudioTrack緩沖區* 3. 更頻繁的數據讀取* 4. 優化文件IO*/
object LowLatencyPullHelper {private const val TAG = "LowLatencyPullHelper"// 音頻參數private const val SAMPLE_RATE = 16000private const val CHANNELS = 1private const val AUDIO_FORMAT = AudioFormat.ENCODING_PCM_16BIT// 低延遲參數private const val SMALL_BUFFER_SIZE = 1024  // 使用更小的緩沖區private const val READ_INTERVAL_MS = 20     // 更頻繁的讀取間隔// 拉流狀態private val isPulling = AtomicBoolean(false)private var ffmpegExecutionId: Long = 0private var audioTrack: AudioTrack? = nullprivate var playbackThread: Thread? = null// 文件讀取位置private val fileReadPosition = AtomicLong(0)// 文件路徑private val cacheDir = File(App.ctx.cacheDir, "low_latency_pull")private val outputPcmFile = File(cacheDir, "realtime_audio.pcm")/*** 開始低延遲拉流播放*/fun startPulling(sdpFilePath: String): Boolean {if (isPulling.get()) {LogUtils.w(TAG, "拉流已在進行中")return false}val sdpFile = File(sdpFilePath)if (!sdpFile.exists()) {LogUtils.e(TAG, "SDP文件不存在: $sdpFilePath")return false}return try {initializeFiles()startLowLatencyDecoding(sdpFilePath)startLowLatencyPlayback()isPulling.set(true)fileReadPosition.set(0)LogUtils.i(TAG, "? 低延遲拉流播放啟動成功")true} catch (e: Exception) {LogUtils.e(TAG, "? 低延遲拉流播放啟動失敗", e)stopPulling()false}}/*** 停止拉流*/fun stopPulling() {if (!isPulling.get()) {return}isPulling.set(false)// 停止FFmpegif (ffmpegExecutionId != 0L) {FFmpeg.cancel(ffmpegExecutionId)ffmpegExecutionId = 0}// 停止音頻播放audioTrack?.stop()audioTrack?.release()audioTrack = null// 停止播放線程playbackThread?.interrupt()playbackThread = nullLogUtils.i(TAG, "🛑 低延遲拉流已停止")}/*** 獲取拉流狀態*/fun isPulling(): Boolean = isPulling.get()/*** 獲取拉流信息*/fun getPullInfo(): String {val fileSize = if (outputPcmFile.exists()) {"${outputPcmFile.length() / 1024}KB"} else {"0KB"}return "拉流狀態: ${if (isPulling.get()) "進行中" else "已停止"}\n" +"解碼文件: ${outputPcmFile.absolutePath}\n" +"文件大小: $fileSize\n" +"讀取位置: ${fileReadPosition.get() / 1024}KB\n" +"優化模式: 低延遲"}/*** 初始化文件和目錄*/private fun initializeFiles() {if (!cacheDir.exists()) {cacheDir.mkdirs()}// 清理舊文件if (outputPcmFile.exists()) {outputPcmFile.delete()}}/*** 啟動低延遲FFmpeg解碼*/private fun startLowLatencyDecoding(sdpFilePath: String) {GlobalScope.launch(Dispatchers.IO) {// 減少等待時間Thread.sleep(500)// 構建超低延遲FFmpeg解碼命令val command = "-protocol_whitelist file,udp,rtp " +"-fflags +nobuffer+flush_packets " +      // 禁用緩沖并立即刷新"-flags low_delay " +                     // 低延遲模式"-probesize 32 " +                        // 最小探測大小"-analyzeduration 0 " +                   // 不分析流"-max_delay 0 " +                         // 最大延遲為0"-reorder_queue_size 0 " +                // 禁用重排序隊列"-rw_timeout 3000000 " +                  // 3秒超時"-i $sdpFilePath " +"-acodec pcm_s16le " +"-ar $SAMPLE_RATE " +"-ac $CHANNELS " +"-f s16le " +"-flush_packets 1 " +                     // 立即刷新數據包"${outputPcmFile.absolutePath}"LogUtils.i(TAG, "低延遲FFmpeg解碼命令: $command")ffmpegExecutionId = FFmpeg.executeAsync(command) { executionId, returnCode ->LogUtils.i(TAG, "FFmpeg解碼結束: executionId=$executionId, returnCode=$returnCode")when (returnCode) {Config.RETURN_CODE_SUCCESS -> {LogUtils.i(TAG, "? 解碼正常結束")}Config.RETURN_CODE_CANCEL -> {LogUtils.i(TAG, "🛑 解碼被用戶取消")}else -> {LogUtils.w(TAG, "?? 解碼異常結束，返回碼: $returnCode")}}}}}/*** 啟動低延遲音頻播放*/private fun startLowLatencyPlayback() {// 使用最小緩沖區val minBufferSize = AudioTrack.getMinBufferSize(SAMPLE_RATE,AudioFormat.CHANNEL_OUT_MONO,AUDIO_FORMAT)// 使用稍大于最小緩沖區的大小，但不要太大val bufferSize = minBufferSize * 2audioTrack = AudioTrack(AudioManager.STREAM_MUSIC,SAMPLE_RATE,AudioFormat.CHANNEL_OUT_MONO,AUDIO_FORMAT,bufferSize,AudioTrack.MODE_STREAM)// 設置低延遲模式（API 26+）try {if (android.os.Build.VERSION.SDK_INT >= android.os.Build.VERSION_CODES.O) {val audioAttributes = android.media.AudioAttributes.Builder().setUsage(android.media.AudioAttributes.USAGE_MEDIA).setContentType(android.media.AudioAttributes.CONTENT_TYPE_MUSIC).setFlags(android.media.AudioAttributes.FLAG_LOW_LATENCY).build()audioTrack = AudioTrack.Builder().setAudioAttributes(audioAttributes).setAudioFormat(AudioFormat.Builder().setEncoding(AUDIO_FORMAT).setSampleRate(SAMPLE_RATE).setChannelMask(AudioFormat.CHANNEL_OUT_MONO).build()).setBufferSizeInBytes(bufferSize).setTransferMode(AudioTrack.MODE_STREAM).build()}} catch (e: Exception) {LogUtils.w(TAG, "無法設置低延遲AudioTrack，使用默認配置", e)}audioTrack?.play()LogUtils.i(TAG, "AudioTrack初始化完成，緩沖區大小: $bufferSize")// 啟動高頻率播放線程playbackThread = Thread {val buffer = ByteArray(SMALL_BUFFER_SIZE)var totalPlayed = 0var lastLogTime = System.currentTimeMillis()LogUtils.i(TAG, "低延遲音頻播放線程啟動")while (isPulling.get() && !Thread.currentThread().isInterrupted) {try {if (outputPcmFile.exists()) {val currentFileSize = outputPcmFile.length()val currentReadPos = fileReadPosition.get()// 如果有新數據可讀if (currentFileSize > currentReadPos) {RandomAccessFile(outputPcmFile, "r").use { randomAccessFile ->randomAccessFile.seek(currentReadPos)val bytesRead = randomAccessFile.read(buffer)if (bytesRead > 0) {audioTrack?.write(buffer, 0, bytesRead)totalPlayed += bytesReadfileReadPosition.addAndGet(bytesRead.toLong())// 每2秒打印一次狀態val currentTime = System.currentTimeMillis()if (currentTime - lastLogTime > 2000) {LogUtils.d(TAG, "🔊 低延遲播放: ${totalPlayed / 1024}KB, 延遲: ${(currentFileSize - currentReadPos) / 32}ms")lastLogTime = currentTime}}}}}// 高頻率檢查，減少延遲Thread.sleep(READ_INTERVAL_MS.toLong())} catch (e: Exception) {LogUtils.e(TAG, "低延遲播放異常", e)Thread.sleep(100)}}LogUtils.i(TAG, "低延遲播放線程結束，總計播放: ${totalPlayed / 1024}KB")}playbackThread?.start()}
}