一、GetX 状态管理的设计
1. 深入理解 ExoPlayer 与状态封装
ExoPlayer 是 Android 平台上强大的媒体播放引擎，它具有丰富的状态和事件回调。在使用 GetX 进行状态管理时，需要深入理解 ExoPlayer 的各种状态，如 STATE_IDLE、STATE_BUFFERING、STATE_READY 和 STATE_ENDED 等，以及播放位置、缓冲位置等信息。
import com.google.android.exoplayer2.ExoPlayer
import com.google.android.exoplayer2.Player
import com.google.android.exoplayer2.source.MediaSource
import com.google.android.exoplayer2.source.ProgressiveMediaSource
import com.google.android.exoplayer2.upstream.DefaultDataSourceFactory
import com.google.android.exoplayer2.util.Util
import io.reactivex.rxjava3.subjects.BehaviorSubject
import io.reactivex.rxjava3.subjects.PublishSubject
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import org.koin.core.component.KoinComponent
import org.koin.core.component.inject
import java.util.concurrent.TimeUnitclass MusicPlayerController : KoinComponent {private val context: android.content.Context by inject()private val player: ExoPlayer = ExoPlayer.Builder(context).build()// 封装播放状态private val _playStateSubject = BehaviorSubject.createDefault<PlayState>(PlayState.IDLE)val playStateFlow: Flow<PlayState> = _playStateSubject.asFlow()// 封装播放位置private val _positionSubject = BehaviorSubject.createDefault(0L)val positionFlow: Flow<Long> = _positionSubject.asFlow()// 封装缓冲位置private val _bufferedPositionSubject = BehaviorSubject.createDefault(0L)val bufferedPositionFlow: Flow<Long> = _bufferedPositionSubject.asFlow()init {player.addListener(object : Player.EventListener {override fun onPlayerStateChanged(playWhenReady: Boolean, playbackState: Int) {val newState = when (playbackState) {Player.STATE_IDLE -> PlayState.IDLEPlayer.STATE_BUFFERING -> PlayState.BUFFERINGPlayer.STATE_READY -> if (playWhenReady) PlayState.PLAYING else PlayState.PAUSEDPlayer.STATE_ENDED -> PlayState.ENDEDelse -> PlayState.IDLE}_playStateSubject.onNext(newState)}override fun onPositionDiscontinuity(reason: Int) {_positionSubject.onNext(player.currentPosition)}override fun onIsPlayingChanged(isPlaying: Boolean) {if (isPlaying) {startPositionUpdates()} else {stopPositionUpdates()}}override fun onBufferingChanged(isBuffering: Boolean) {if (isBuffering) {_bufferedPositionSubject.onNext(player.bufferedPosition)}}})}private var positionUpdateJob: Job? = nullprivate fun startPositionUpdates() {positionUpdateJob = GlobalScope.launch {while (isActive) {_positionSubject.onNext(player.currentPosition)delay(100) // 每 100ms 更新一次位置}}}private fun stopPositionUpdates() {positionUpdateJob?.cancel()}fun play(url: String) {val dataSourceFactory = DefaultDataSourceFactory(context, Util.getUserAgent(context, "MusicPlayer"))val mediaSource: MediaSource = ProgressiveMediaSource.Factory(dataSourceFactory).createMediaSource(android.net.Uri.parse(url))player.setMediaSource(mediaSource)player.prepare()player.play()}fun pause() {player.pause()}fun seekTo(position: Long) {player.seekTo(position)}fun release() {player.release()}
}enum class PlayState {IDLE, BUFFERING, PLAYING, PAUSED, ENDED
}在上述代码中，我们创建了一个 MusicPlayerController 类，它继承自 KoinComponent 以便使用依赖注入。通过 BehaviorSubject 封装了播放状态、播放位置和缓冲位置，这样可以方便地将这些状态暴露为 Flow，供 UI 层订阅。在 init 块中，我们为 ExoPlayer 添加了事件监听器，根据不同的状态更新相应的 Subject。同时，为了实时更新播放位置，我们使用协程每隔 100ms 更新一次位置。
2. 跨页面状态共享与 UI 自动更新
GetX 的 Rx 响应式变量使得状态的变化能够自动通知到订阅的 UI 组件。在 Flutter 中，我们可以使用 Obx 或 GetX 组件来监听状态的变化。
import 'package:flutter/material.dart';
import 'package:get/get.dart';
import 'package:your_app/music_player_controller.dart';class MusicPlayerPage extends StatelessWidget {final MusicPlayerController controller = Get.put(MusicPlayerController());@overrideWidget build(BuildContext context) {return Scaffold(appBar: AppBar(title: Text('Music Player'),),body: Column(mainAxisAlignment: MainAxisAlignment.center,children: [Obx(() => Text('Play State: ${controller.playStateFlow.value}')),Obx(() => Text('Position: ${controller.positionFlow.value} ms')),Obx(() => Text('Buffered Position: ${controller.bufferedPositionFlow.value} ms')),ElevatedButton(onPressed: () => controller.play('your_music_url'),child: Text('Play'),),ElevatedButton(onPressed: () => controller.pause(),child: Text('Pause'),),ElevatedButton(onPressed: () => controller.seekTo(5000), // 跳到 5s 位置child: Text('Seek to 5s'),),],),);}
}在这个 Flutter 页面中，我们使用 Get.put 方法将 MusicPlayerController 实例化并注入到 GetX 管理中。通过 Obx 组件监听 playStateFlow、positionFlow 和 bufferedPositionFlow 的变化，当这些状态发生变化时，UI 会自动更新。
3. 处理多端播放进度精准对齐
多端播放进度精准对齐是一个复杂的问题，主要难点在于不同设备的时间戳可能不一致，网络延迟也会影响同步的准确性。我们可以通过以下步骤来解决：
- 使用 WebSocket 实时推送播放事件：当用户在某一端进行播放操作（如播放、暂停、Seek 等）时，该端将操作事件和对应的时间戳通过 WebSocket 发送到服务器。
- 服务器广播事件：服务器接收到事件后，将其广播给所有连接的客户端。
- 客户端接收事件并调整进度：客户端接收到事件后，根据事件中的时间戳和本地时间戳计算时间差，然后调整本地播放进度。
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import okhttp3.*
import java.io.IOException
import java.util.concurrent.TimeUnitclass WebSocketManager {private val client = OkHttpClient.Builder().pingInterval(10, TimeUnit.SECONDS).build()private var webSocket: WebSocket? = nullprivate val eventFlow = MutableSharedFlow<PlayEvent>()fun connect(url: String) {val request = Request.Builder().url(url).build()webSocket = client.newWebSocket(request, object : WebSocketListener() {override fun onOpen(webSocket: WebSocket, response: Response) {println("WebSocket connected")}override fun onMessage(webSocket: WebSocket, text: String) {val event = PlayEvent.fromJson(text)GlobalScope.launch {eventFlow.emit(event)}}override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {println("WebSocket failure: ${t.message}")reconnect(url)}override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {println("WebSocket closed: $code - $reason")reconnect(url)}})}private fun reconnect(url: String) {GlobalScope.launch {delay(5000) // 5s 后重试connect(url)}}fun sendEvent(event: PlayEvent) {val json = event.toJson()webSocket?.send(json)}fun observeEvents(): Flow<PlayEvent> = eventFlow
}data class PlayEvent(val eventType: EventType,val position: Long,val timestamp: Long
) {enum class EventType {PLAY, PAUSE, SEEK}fun toJson(): String {// 实现 JSON 序列化return ""}companion object {fun fromJson(json: String): PlayEvent {// 实现 JSON 反序列化return PlayEvent(EventType.PLAY, 0, 0)}}
}在上述代码中，我们创建了一个 WebSocketManager 类，用于管理 WebSocket 连接。通过 connect 方法连接到服务器，当接收到消息时，将其解析为 PlayEvent 并通过 eventFlow 发射出去。如果连接失败或关闭，会在 5s 后重试连接。同时，提供了 sendEvent 方法用于发送播放事件。
二、网络与 WebSocket 的结合
1. 架构设计
HTTP 请求用于初始化数据，如歌曲列表、用户信息等。这些数据通常是静态的或不经常变化的，使用 HTTP 请求可以利用其成熟的缓存机制和错误处理机制。WebSocket 则负责实时同步，如其他设备切换歌曲、进度更新等。这种架构设计可以充分发挥两种协议的优势，提高系统的性能和实时性。
import okhttp3.*
import java.io.IOExceptionclass HttpManager {private val client = OkHttpClient()fun getSongList(url: String, callback: Callback) {val request = Request.Builder().url(url).build()client.newCall(request).enqueue(callback)}fun getUserInfo(url: String, callback: Callback) {val request = Request.Builder().url(url).build()client.newCall(request).enqueue(callback)}
}在这个 HttpManager 类中，我们使用 OkHttp 库来处理 HTTP 请求。通过 getSongList 和 getUserInfo 方法分别获取歌曲列表和用户信息，使用 enqueue 方法进行异步请求。
2. Kotlin 协程优化
Kotlin 协程可以方便地处理异步操作，避免阻塞主线程。在处理 WebSocket 心跳、数据解析等耗时操作时，我们可以使用 withContext(Dispatchers.IO) 来切换到 IO 线程。
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import okhttp3.*
import java.io.IOException
import java.util.concurrent.TimeUnitclass WebSocketManager {private val client = OkHttpClient.Builder().pingInterval(10, TimeUnit.SECONDS).build()private var webSocket: WebSocket? = nullprivate val eventFlow = MutableSharedFlow<PlayEvent>()suspend fun connect(url: String) = withContext(Dispatchers.IO) {val request = Request.Builder().url(url).build()webSocket = client.newWebSocket(request, object : WebSocketListener() {override fun onOpen(webSocket: WebSocket, response: Response) {println("WebSocket connected")}override fun onMessage(webSocket: WebSocket, text: String) {val event = PlayEvent.fromJson(text)GlobalScope.launch {eventFlow.emit(event)}}override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {println("WebSocket failure: ${t.message}")GlobalScope.launch {reconnect(url)}}override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {println("WebSocket closed: $code - $reason")GlobalScope.launch {reconnect(url)}}})}private suspend fun reconnect(url: String) = withContext(Dispatchers.IO) {delay(5000) // 5s 后重试connect(url)}fun sendEvent(event: PlayEvent) {val json = event.toJson()webSocket?.send(json)}fun observeEvents(): Flow<PlayEvent> = eventFlow
}在这个改进后的 WebSocketManager 类中，connect 和 reconnect 方法都使用了 withContext(Dispatchers.IO) 来切换到 IO 线程，避免阻塞主线程。同时，使用 MutableSharedFlow 来处理事件的发射和订阅。
3. Flow 处理数据流与 UI 映射
利用 Flow 可以方便地处理数据流，通过 collectAsState() 将实时数据映射到 UI 层。
import 'package:flutter/material.dart';
import 'package:get/get.dart';
import 'package:your_app/music_player_controller.dart';
import 'package:your_app/websocket_manager.dart';class MusicPlayerPage extends StatelessWidget {final MusicPlayerController controller = Get.put(MusicPlayerController());final WebSocketManager webSocketManager = Get.put(WebSocketManager());@overrideWidget build(BuildContext context) {return Scaffold(appBar: AppBar(title: Text('Music Player'),),body: Column(mainAxisAlignment: MainAxisAlignment.center,children: [Obx(() => Text('Play State: ${controller.playStateFlow.value}')),Obx(() => Text('Position: ${controller.positionFlow.value} ms')),Obx(() => Text('Buffered Position: ${controller.bufferedPositionFlow.value} ms')),ElevatedButton(onPressed: () => controller.play('your_music_url'),child: Text('Play'),),ElevatedButton(onPressed: () => controller.pause(),child: Text('Pause'),),ElevatedButton(onPressed: () => controller.seekTo(5000), // 跳到 5s 位置child: Text('Seek to 5s'),),StreamBuilder<PlayEvent>(stream: webSocketManager.observeEvents().asStream(),builder: (context, snapshot) {if (snapshot.hasData) {return Text('Received event: ${snapshot.data?.eventType} at ${snapshot.data?.position} ms');} else {return Text('No event received');}},),],),);}
}在这个 Flutter 页面中，我们使用 StreamBuilder 来监听 WebSocketManager 的 eventFlow，当接收到新的事件时，更新 UI 显示事件信息。
4. 网络波动处理
当检测到网络波动时，协程自动重试 WebSocket 连接，并缓存未同步的播放事件，网络恢复后批量补发。
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import okhttp3.*
import java.io.IOException
import java.util.concurrent.TimeUnitclass WebSocketManager {private val client = OkHttpClient.Builder().pingInterval(10, TimeUnit.SECONDS).build()private var webSocket: WebSocket? = nullprivate val eventFlow = MutableSharedFlow<PlayEvent>()private val pendingEvents = mutableListOf<PlayEvent>()suspend fun connect(url: String) = withContext(Dispatchers.IO) {val request = Request.Builder().url(url).build()webSocket = client.newWebSocket(request, object : WebSocketListener() {override fun onOpen(webSocket: WebSocket, response: Response) {println("WebSocket connected")sendPendingEvents()}override fun onMessage(webSocket: WebSocket, text: String) {val event = PlayEvent.fromJson(text)GlobalScope.launch {eventFlow.emit(event)}}override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {println("WebSocket failure: ${t.message}")GlobalScope.launch {reconnect(url)}}override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {println("WebSocket closed: $code - $reason")GlobalScope.launch {reconnect(url)}}})}private suspend fun reconnect(url: String) = withContext(Dispatchers.IO) {delay(5000) // 5s 后重试connect(url)}fun sendEvent(event: PlayEvent) {if (webSocket?.connectionState() == WebSocket.State.OPEN) {val json = event.toJson()webSocket?.send(json)} else {pendingEvents.add(event)}}private fun sendPendingEvents() {pendingEvents.forEach { event ->val json = event.toJson()webSocket?.send(json)}pendingEvents.clear()}fun observeEvents(): Flow<PlayEvent> = eventFlow
} 在这个改进后的 WebSocketManager 类中，我们添加了一个 pendingEvents 列表来缓存未同步的播放事件。当连接成功时，调用 sendPendingEvents 方法批量发送这些事件。在 sendEvent 方法中，如果连接未打开，则将事件添加到 pendingEvents 列表中。
感谢观看！！！