Android应用多数据源事件聚合器实现：优先处理HTTP事件

这个需求挺典型的，我之前在做Android实时数据聚合时也遇到过类似场景，结合Android的线程模型和组件特性，给你一套清晰的实现方案：

我们需要一个单例模式的事件聚合器，它负责统一接收两类数据源的事件，核心目标是保证HTTP事件的处理优先级绝对高于非HTTP事件。实现的关键是用队列（或优先级队列）管理事件顺序，同时用独立工作线程串行处理事件，避免并发处理带来的业务逻辑混乱。

1. 定义统一的Event实体类 #

先确保两类数据源的事件都能映射到同一个实体类，比如：

data class Event(
    val id: String,
    val content: String,
    // 其他业务字段（如时间戳、类型标识等）
)

2. 实现事件聚合器（单例） #

这里提供两种可选方案，你可以根据业务扩展性需求选择：

方案一：双队列+优先遍历HTTP队列 #

逻辑直观，适合仅区分两类事件的场景，严格保证HTTP事件全部处理完成后才会处理非HTTP事件：

class EventAggregator private constructor() {
    // HTTP事件队列（优先处理）
    private val httpEventQueue = LinkedBlockingQueue<Event>()
    // 非HTTP事件队列
    private val nonHttpEventQueue = LinkedBlockingQueue<Event>()
    // 工作线程，负责循环处理事件
    private val workerThread = Thread(this::processEventsLoop).apply { start() }

    // 事件处理的核心业务逻辑，你可以根据需求自行扩展
    private fun processEvent(event: Event) {
        Log.d("EventAggregator", "Processing event: ${event.id}")
        // 比如：存储到本地数据库、发送UI通知、更新内存缓存等
    }

    // 事件循环处理逻辑
    private fun processEventsLoop() {
        while (!Thread.currentThread().isInterrupted) {
            try {
                // Step 1: 优先处理所有HTTP事件，直到队列清空
                var event = httpEventQueue.poll()
                while (event != null) {
                    processEvent(event)
                    event = httpEventQueue.poll()
                }
                // Step 2: HTTP队列为空时，才处理非HTTP事件（阻塞等待新事件）
                event = nonHttpEventQueue.take()
                processEvent(event)
            } catch (e: InterruptedException) {
                // 线程被中断，退出循环
                Thread.currentThread().interrupt()
            } catch (e: Exception) {
                // 单个事件处理失败时，捕获异常避免整个工作线程崩溃
                Log.e("EventAggregator", "Failed to process event", e)
            }
        }
    }

    // 对外提供的方法：批量提交HTTP事件
    fun submitHttpEvents(events: List<Event>) {
        httpEventQueue.addAll(events)
    }

    // 对外提供的方法：单个提交非HTTP事件
    fun submitNonHttpEvent(event: Event) {
        nonHttpEventQueue.add(event)
    }

    // 单例实现，保证全局唯一实例
    companion object {
        val instance: EventAggregator by lazy { EventAggregator() }
    }
}

方案二：优先级队列（更灵活） #

如果未来可能扩展更多优先级的事件类型，推荐用这种方案，通过给事件标记优先级自动排序：

// 包装Event，添加优先级属性（也可以直接给Event类加优先级字段）
data class EventWrapper(
    val event: Event,
    // 优先级规则：数值越小，优先级越高（HTTP事件设为1，非HTTP设为2）
    val priority: Int
)

class EventAggregator private constructor() {
    // 优先级队列，自动按priority升序排序
    private val eventQueue = PriorityBlockingQueue<EventWrapper>(
        10,
        Comparator.comparingInt { it.priority }
    )
    private val workerThread = Thread(this::processEventsLoop).apply { start() }

    private fun processEvent(event: Event) {
        // 你的事件处理逻辑
        Log.d("EventAggregator", "Processing event: ${event.id}")
    }

    private fun processEventsLoop() {
        while (!Thread.currentThread().isInterrupted) {
            try {
                // 阻塞取出当前优先级最高的事件
                val wrapper = eventQueue.take()
                processEvent(wrapper.event)
            } catch (e: InterruptedException) {
                Thread.currentThread().interrupt()
            } catch (e: Exception) {
                Log.e("EventAggregator", "Failed to process event", e)
            }
        }
    }

    fun submitHttpEvents(events: List<Event>) {
        events.forEach {
            eventQueue.add(EventWrapper(it, priority = 1))
        }
    }

    fun submitNonHttpEvent(event: Event) {
        eventQueue.add(EventWrapper(event, priority = 2))
    }

    companion object {
        val instance: EventAggregator by lazy { EventAggregator() }
    }
}

3. 与现有组件集成 #

非HTTP事件（Android Service） #

在你已有的处理非HTTP事件的Service中，每当接收到单个事件时，直接调用聚合器的提交方法：

// 假设Service中收到非HTTP事件的回调方法
fun onNonHttpEventReceived(event: Event) {
    EventAggregator.instance.submitNonHttpEvent(event)
}

HTTP事件（周期性调用） #

不管你用Retrofit、OkHttp还是其他HTTP库，在请求成功拿到events数组后，提交到聚合器即可：

// 示例：用Coroutine发起HTTP请求（也可以用回调方式）
scope.launch(Dispatchers.IO) {
    try {
        val events = apiService.fetchEvents() // 你的HTTP接口调用
        EventAggregator.instance.submitHttpEvents(events)
    } catch (e: Exception) {
        // 处理HTTP请求失败的情况
        Log.e("HttpEventFetcher", "Failed to fetch HTTP events", e)
    }
}

4. 额外注意事项 #

• 线程生命周期管理：在Application的onCreate方法中初始化聚合器，在App退出时（比如通过全局退出逻辑）中断工作线程，避免内存泄漏。
• 后台Service限制：如果你的非HTTP Service是后台Service，在Android 8.0+系统中需要升级为Foreground Service，避免被系统杀死。
• 事件去重：如果两类数据源可能产生重复Event，可以在聚合器中添加去重逻辑（比如维护一个最近处理的Event ID集合）。
• 性能优化：如果事件量极大，可以考虑用多线程池，但要给HTTP事件分配更高优先级的线程；一般场景下单线程串行处理足够，能避免并发问题。

内容的提问来源于stack exchange，提问作者Ildar Zaripov