项目总结：GetX + Kotlin 协程实现跨端音乐播放实时同步
一、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.TimeUnit
class 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.IDLE
Player.STATE_BUFFERING -> PlayState.BUFFERING
Player.STATE_READY -> if (playWhenReady) PlayState.PLAYING else PlayState.PAUSED
Player.STATE_ENDED -> PlayState.ENDED
else -> 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? = null
private 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());
@override
Widget 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.TimeUnit
class WebSocketManager {
private val client = OkHttpClient.Builder()
.pingInterval(10, TimeUnit.SECONDS)
.build()
private var webSocket: WebSocket? = null
private 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.IOException
class 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.TimeUnit
class WebSocketManager {
private val client = OkHttpClient.Builder()
.pingInterval(10, TimeUnit.SECONDS)
.build()
private var webSocket: WebSocket? = null
private 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());
@override
Widget 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.TimeUnit
class WebSocketManager {
private val client = OkHttpClient.Builder()
.pingInterval(10, TimeUnit.SECONDS)
.build()
private var webSocket: WebSocket? = null
private 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 列表中。
感谢观看！！！
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]