项目总结：GetX + Kotlin 协程实现跨端音乐播放实时同步 ...

一、GetX 状态管理的设计

1. 深入明白 ExoPlayer 与状态封装

ExoPlayer 是 Android 平台上强盛的媒体播放引擎，它具有丰富的状态和事件回调。在使用 GetX 进行状态管理时，必要深入明白 ExoPlayer 的各种状态，如 STATE_IDLE、STATE_BUFFERING、STATE_READY 和 STATE_ENDED 等，以及播放位置、缓冲位置等信息。

1. import com.google.android.exoplayer2.ExoPlayer
2. import com.google.android.exoplayer2.Player
3. import com.google.android.exoplayer2.source.MediaSource
4. import com.google.android.exoplayer2.source.ProgressiveMediaSource
5. import com.google.android.exoplayer2.upstream.DefaultDataSourceFactory
6. import com.google.android.exoplayer2.util.Util
7. import io.reactivex.rxjava3.subjects.BehaviorSubject
8. import io.reactivex.rxjava3.subjects.PublishSubject
9. import kotlinx.coroutines.*
10. import kotlinx.coroutines.flow.*
11. import org.koin.core.component.KoinComponent
12. import org.koin.core.component.inject
13. import java.util.concurrent.TimeUnit
15. class MusicPlayerController : KoinComponent {
16.     private val context: android.content.Context by inject()
17.     private val player: ExoPlayer = ExoPlayer.Builder(context).build()
19.     // 封装播放状态
20.     private val _playStateSubject = BehaviorSubject.createDefault<PlayState>(PlayState.IDLE)
21.     val playStateFlow: Flow<PlayState> = _playStateSubject.asFlow()
23.     // 封装播放位置
24.     private val _positionSubject = BehaviorSubject.createDefault(0L)
25.     val positionFlow: Flow<Long> = _positionSubject.asFlow()
27.     // 封装缓冲位置
28.     private val _bufferedPositionSubject = BehaviorSubject.createDefault(0L)
29.     val bufferedPositionFlow: Flow<Long> = _bufferedPositionSubject.asFlow()
31.     init {
32.         player.addListener(object : Player.EventListener {
33.             override fun onPlayerStateChanged(playWhenReady: Boolean, playbackState: Int) {
34.                 val newState = when (playbackState) {
35.                     Player.STATE_IDLE -> PlayState.IDLE
36.                     Player.STATE_BUFFERING -> PlayState.BUFFERING
37.                     Player.STATE_READY -> if (playWhenReady) PlayState.PLAYING else PlayState.PAUSED
38.                     Player.STATE_ENDED -> PlayState.ENDED
39.                     else -> PlayState.IDLE
40.                 }
41.                 _playStateSubject.onNext(newState)
42.             }
44.             override fun onPositionDiscontinuity(reason: Int) {
45.                 _positionSubject.onNext(player.currentPosition)
46.             }
48.             override fun onIsPlayingChanged(isPlaying: Boolean) {
49.                 if (isPlaying) {
50.                     startPositionUpdates()
51.                 } else {
52.                     stopPositionUpdates()
53.                 }
54.             }
56.             override fun onBufferingChanged(isBuffering: Boolean) {
57.                 if (isBuffering) {
58.                     _bufferedPositionSubject.onNext(player.bufferedPosition)
59.                 }
60.             }
61.         })
62.     }
64.     private var positionUpdateJob: Job? = null
65.     private fun startPositionUpdates() {
66.         positionUpdateJob = GlobalScope.launch {
67.             while (isActive) {
68.                 _positionSubject.onNext(player.currentPosition)
69.                 delay(100) // 每 100ms 更新一次位置
70.             }
71.         }
72.     }
74.     private fun stopPositionUpdates() {
75.         positionUpdateJob?.cancel()
76.     }
78.     fun play(url: String) {
79.         val dataSourceFactory = DefaultDataSourceFactory(context, Util.getUserAgent(context, "MusicPlayer"))
80.         val mediaSource: MediaSource = ProgressiveMediaSource.Factory(dataSourceFactory)
81.            .createMediaSource(android.net.Uri.parse(url))
82.         player.setMediaSource(mediaSource)
83.         player.prepare()
84.         player.play()
85.     }
87.     fun pause() {
88.         player.pause()
89.     }
91.     fun seekTo(position: Long) {
92.         player.seekTo(position)
93.     }
95.     fun release() {
96.         player.release()
97.     }
98. }
100. enum class PlayState {
101.     IDLE, BUFFERING, PLAYING, PAUSED, ENDED
102. }

复制代码

在上述代码中，我们创建了一个 MusicPlayerController 类，它继承自 KoinComponent 以便使用依赖注入。通过 BehaviorSubject 封装了播放状态、播放位置和缓冲位置，如允许以方便地将这些状态袒露为 Flow，供 UI 层订阅。在 init 块中，我们为 ExoPlayer 添加了事件监听器，根据不同的状态更新相应的 Subject。同时，为了实时更新播放位置，我们使用协程每隔 100ms 更新一次位置。
2. 跨页面状态共享与 UI 自动更新

GetX 的 Rx 响应式变量使得状态的变革能够自动通知到订阅的 UI 组件。在 Flutter 中，我们可以使用 Obx 或 GetX 组件来监听状态的变革。

1. import 'package:flutter/material.dart';
2. import 'package:get/get.dart';
3. import 'package:your_app/music_player_controller.dart';
5. class MusicPlayerPage extends StatelessWidget {
6.   final MusicPlayerController controller = Get.put(MusicPlayerController());
8.   @override
9.   Widget build(BuildContext context) {
10.     return Scaffold(
11.       appBar: AppBar(
12.         title: Text('Music Player'),
13.       ),
14.       body: Column(
15.         mainAxisAlignment: MainAxisAlignment.center,
16.         children: [
17.           Obx(() => Text('Play State: ${controller.playStateFlow.value}')),
18.           Obx(() => Text('Position: ${controller.positionFlow.value} ms')),
19.           Obx(() => Text('Buffered Position: ${controller.bufferedPositionFlow.value} ms')),
20.           ElevatedButton(
21.             onPressed: () => controller.play('your_music_url'),
22.             child: Text('Play'),
23.           ),
24.           ElevatedButton(
25.             onPressed: () => controller.pause(),
26.             child: Text('Pause'),
27.           ),
28.           ElevatedButton(
29.             onPressed: () => controller.seekTo(5000), // 跳到 5s 位置
30.             child: Text('Seek to 5s'),
31.           ),
32.         ],
33.       ),
34.     );
35.   }
36. }

复制代码

在这个 Flutter 页面中，我们使用 Get.put 方法将 MusicPlayerController 实例化并注入到 GetX 管理中。通过 Obx 组件监听 playStateFlow、positionFlow 和 bufferedPositionFlow 的变革，当这些状态发生变革时，UI 会自动更新。
3. 处理多端播放进度精准对齐

多端播放进度精准对齐是一个复杂的题目，主要难点在于不同装备的时间戳可能不一致，网络延迟也会影响同步的准确性。我们可以通过以下步骤来解决：

• 使用 WebSocket 实时推送播放事件：当用户在某一端进行播放操纵（如播放、暂停、Seek 等）时，该端将操纵事件和对应的时间戳通过 WebSocket 发送到服务器。
  
• 服务器广播事件：服务器接收到事件后，将其广播给全部毗连的客户端。
  
• 客户端接收事件并调解进度：客户端接收到事件后，根据事件中的时间戳和本地时间戳盘算时间差，然后调解本地播放进度。
  

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*
3. import okhttp3.*
4. import java.io.IOException
5. import java.util.concurrent.TimeUnit
7. class WebSocketManager {
8.     private val client = OkHttpClient.Builder()
9.        .pingInterval(10, TimeUnit.SECONDS)
10.        .build()
11.     private var webSocket: WebSocket? = null
12.     private val eventFlow = MutableSharedFlow<PlayEvent>()
14.     fun connect(url: String) {
15.         val request = Request.Builder()
16.            .url(url)
17.            .build()
18.         webSocket = client.newWebSocket(request, object : WebSocketListener() {
19.             override fun onOpen(webSocket: WebSocket, response: Response) {
20.                 println("WebSocket connected")
21.             }
23.             override fun onMessage(webSocket: WebSocket, text: String) {
24.                 val event = PlayEvent.fromJson(text)
25.                 GlobalScope.launch {
26.                     eventFlow.emit(event)
27.                 }
28.             }
30.             override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
31.                 println("WebSocket failure: ${t.message}")
32.                 reconnect(url)
33.             }
35.             override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
36.                 println("WebSocket closed: $code - $reason")
37.                 reconnect(url)
38.             }
39.         })
40.     }
42.     private fun reconnect(url: String) {
43.         GlobalScope.launch {
44.             delay(5000) // 5s 后重试
45.             connect(url)
46.         }
47.     }
49.     fun sendEvent(event: PlayEvent) {
50.         val json = event.toJson()
51.         webSocket?.send(json)
52.     }
54.     fun observeEvents(): Flow<PlayEvent> = eventFlow
55. }
57. data class PlayEvent(
58.     val eventType: EventType,
59.     val position: Long,
60.     val timestamp: Long
61. ) {
62.     enum class EventType {
63.         PLAY, PAUSE, SEEK
64.     }
66.     fun toJson(): String {
67.         // 实现 JSON 序列化
68.         return ""
69.     }
71.     companion object {
72.         fun fromJson(json: String): PlayEvent {
73.             // 实现 JSON 反序列化
74.             return PlayEvent(EventType.PLAY, 0, 0)
75.         }
76.     }
77. }

复制代码

在上述代码中，我们创建了一个 WebSocketManager 类，用于管理 WebSocket 毗连。通过 connect 方法毗连到服务器，当接收到消息时，将其剖析为 PlayEvent 并通过 eventFlow 发射出去。如果毗连失败或关闭，会在 5s 后重试毗连。同时，提供了 sendEvent 方法用于发送播放事件。
二、网络与 WebSocket 的结合

1. 架构设计

HTTP 哀求用于初始化数据，如歌曲列表、用户信息等。这些数据通常是静态的或不常常变革的，使用 HTTP 哀求可以利用其成熟的缓存机制和错误处理机制。WebSocket 则负责实时同步，如其他装备切换歌曲、进度更新等。这种架构设计可以充实发挥两种协议的优势，提高体系的性能和实时性。

1. import okhttp3.*
2. import java.io.IOException
4. class HttpManager {
5.     private val client = OkHttpClient()
7.     fun getSongList(url: String, callback: Callback) {
8.         val request = Request.Builder()
9.            .url(url)
10.            .build()
11.         client.newCall(request).enqueue(callback)
12.     }
14.     fun getUserInfo(url: String, callback: Callback) {
15.         val request = Request.Builder()
16.            .url(url)
17.            .build()
18.         client.newCall(request).enqueue(callback)
19.     }
20. }

复制代码

在这个 HttpManager 类中，我们使用 OkHttp 库来处理 HTTP 哀求。通过 getSongList 和 getUserInfo 方法分别获取歌曲列表和用户信息，使用 enqueue 方法进行异步哀求。
2. Kotlin 协程优化

Kotlin 协程可以方便地处理异步操纵，制止阻塞主线程。在处理 WebSocket 心跳、数据剖析等耗时操纵时，我们可以使用 withContext(Dispatchers.IO) 来切换到 IO 线程。

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*
3. import okhttp3.*
4. import java.io.IOException
5. import java.util.concurrent.TimeUnit
7. class WebSocketManager {
8.     private val client = OkHttpClient.Builder()
9.        .pingInterval(10, TimeUnit.SECONDS)
10.        .build()
11.     private var webSocket: WebSocket? = null
12.     private val eventFlow = MutableSharedFlow<PlayEvent>()
14.     suspend fun connect(url: String) = withContext(Dispatchers.IO) {
15.         val request = Request.Builder()
16.            .url(url)
17.            .build()
18.         webSocket = client.newWebSocket(request, object : WebSocketListener() {
19.             override fun onOpen(webSocket: WebSocket, response: Response) {
20.                 println("WebSocket connected")
21.             }
23.             override fun onMessage(webSocket: WebSocket, text: String) {
24.                 val event = PlayEvent.fromJson(text)
25.                 GlobalScope.launch {
26.                     eventFlow.emit(event)
27.                 }
28.             }
30.             override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
31.                 println("WebSocket failure: ${t.message}")
32.                 GlobalScope.launch {
33.                     reconnect(url)
34.                 }
35.             }
37.             override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
38.                 println("WebSocket closed: $code - $reason")
39.                 GlobalScope.launch {
40.                     reconnect(url)
41.                 }
42.             }
43.         })
44.     }
46.     private suspend fun reconnect(url: String) = withContext(Dispatchers.IO) {
47.         delay(5000) // 5s 后重试
48.         connect(url)
49.     }
51.     fun sendEvent(event: PlayEvent) {
52.         val json = event.toJson()
53.         webSocket?.send(json)
54.     }
56.     fun observeEvents(): Flow<PlayEvent> = eventFlow
57. }

复制代码

在这个改进后的 WebSocketManager 类中，connect 和 reconnect 方法都使用了 withContext(Dispatchers.IO) 来切换到 IO 线程，制止阻塞主线程。同时，使用 MutableSharedFlow 来处理事件的发射和订阅。
3. Flow 处理数据流与 UI 映射

利用 Flow 可以方便地处理数据流，通过 collectAsState() 将实时数据映射到 UI 层。

1. import 'package:flutter/material.dart';
2. import 'package:get/get.dart';
3. import 'package:your_app/music_player_controller.dart';
4. import 'package:your_app/websocket_manager.dart';
6. class MusicPlayerPage extends StatelessWidget {
7.   final MusicPlayerController controller = Get.put(MusicPlayerController());
8.   final WebSocketManager webSocketManager = Get.put(WebSocketManager());
10.   @override
11.   Widget build(BuildContext context) {
12.     return Scaffold(
13.       appBar: AppBar(
14.         title: Text('Music Player'),
15.       ),
16.       body: Column(
17.         mainAxisAlignment: MainAxisAlignment.center,
18.         children: [
19.           Obx(() => Text('Play State: ${controller.playStateFlow.value}')),
20.           Obx(() => Text('Position: ${controller.positionFlow.value} ms')),
21.           Obx(() => Text('Buffered Position: ${controller.bufferedPositionFlow.value} ms')),
22.           ElevatedButton(
23.             onPressed: () => controller.play('your_music_url'),
24.             child: Text('Play'),
25.           ),
26.           ElevatedButton(
27.             onPressed: () => controller.pause(),
28.             child: Text('Pause'),
29.           ),
30.           ElevatedButton(
31.             onPressed: () => controller.seekTo(5000), // 跳到 5s 位置
32.             child: Text('Seek to 5s'),
33.           ),
34.           StreamBuilder<PlayEvent>(
35.             stream: webSocketManager.observeEvents().asStream(),
36.             builder: (context, snapshot) {
37.               if (snapshot.hasData) {
38.                 return Text('Received event: ${snapshot.data?.eventType} at ${snapshot.data?.position} ms');
39.               } else {
40.                 return Text('No event received');
41.               }
42.             },
43.           ),
44.         ],
45.       ),
46.     );
47.   }
48. }

复制代码

在这个 Flutter 页面中，我们使用 StreamBuilder 来监听 WebSocketManager 的 eventFlow，当接收到新的事件时，更新 UI 显示事件信息。
4. 网络波动处理

当检测到网络波动时，协程自动重试 WebSocket 毗连，并缓存未同步的播放事件，网络恢复后批量补发。
 

1. import kotlinx.coroutines.*
2. import kotlinx.coroutines.flow.*
3. import okhttp3.*
4. import java.io.IOException
5. import java.util.concurrent.TimeUnit
7. class WebSocketManager {
8.     private val client = OkHttpClient.Builder()
9.        .pingInterval(10, TimeUnit.SECONDS)
10.        .build()
11.     private var webSocket: WebSocket? = null
12.     private val eventFlow = MutableSharedFlow<PlayEvent>()
13.     private val pendingEvents = mutableListOf<PlayEvent>()
15.     suspend fun connect(url: String) = withContext(Dispatchers.IO) {
16.         val request = Request.Builder()
17.            .url(url)
18.            .build()
19.         webSocket = client.newWebSocket(request, object : WebSocketListener() {
20.             override fun onOpen(webSocket: WebSocket, response: Response) {
21.                 println("WebSocket connected")
22.                 sendPendingEvents()
23.             }
25.             override fun onMessage(webSocket: WebSocket, text: String) {
26.                 val event = PlayEvent.fromJson(text)
27.                 GlobalScope.launch {
28.                     eventFlow.emit(event)
29.                 }
30.             }
32.             override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
33.                 println("WebSocket failure: ${t.message}")
34.                 GlobalScope.launch {
35.                     reconnect(url)
36.                 }
37.             }
39.             override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
40.                 println("WebSocket closed: $code - $reason")
41.                 GlobalScope.launch {
42.                     reconnect(url)
43.                 }
44.             }
45.         })
46.     }
48.     private suspend fun reconnect(url: String) = withContext(Dispatchers.IO) {
49.         delay(5000) // 5s 后重试
50.         connect(url)
51.     }
53.     fun sendEvent(event: PlayEvent) {
54.         if (webSocket?.connectionState() == WebSocket.State.OPEN) {
55.             val json = event.toJson()
56.             webSocket?.send(json)
57.         } else {
58.             pendingEvents.add(event)
59.         }
60.     }
62.     private fun sendPendingEvents() {
63.         pendingEvents.forEach { event ->
64.             val json = event.toJson()
65.             webSocket?.send(json)
66.         }
67.         pendingEvents.clear()
68.     }
70.     fun observeEvents(): Flow<PlayEvent> = eventFlow
71. }

复制代码

 在这个改进后的 WebSocketManager 类中，我们添加了一个 pendingEvents 列表来缓存未同步的播放事件。当毗连乐成时，调用 sendPendingEvents 方法批量发送这些事件。在 sendEvent 方法中，如果毗连未打开，则将事件添加到 pendingEvents 列表中。

感谢观看！！！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。