如何将okhttp-eventsource的BackgroundEventHandler值传递到主线程？

问题原因 #

你调用bes.getEventSource().messages().forEach()时触发NetworkOnMainThreadException，是因为messages()返回的是阻塞式可迭代序列——主线程执行forEach会被阻塞等待新消息，而Android严格禁止主线程执行网络/阻塞操作，因此抛出异常。

解决方案 #

直接在BackgroundEventHandler的onMessage回调中，将消息切换到主线程处理，无需调用messages()方法。以下是两种可靠实现方式：

方式1：使用主线程Handler #

修改BackgroundEventHandler，通过Handler将任务投递到主线程：

public class BackgroundEventHandler implements EventHandler {
    private static final String TAG = "BackgroundEventHandler";
    private final Handler mainHandler;

    // 传入主线程Handler
    public BackgroundEventHandler(Handler mainHandler) {
        this.mainHandler = mainHandler;
    }

    @Override
    public void onMessage(String event, MessageEvent messageEvent) {
        Log.d(TAG, messageEvent.getData().toString());
        // 将消息处理逻辑投递到主线程
        mainHandler.post(() -> {
            // 在这里执行主线程操作：解析JSON、更新UI等
            String jsonData = messageEvent.getData();
            Log.d("MainThread", "Received SSE message: " + jsonData);
            // 示例：解析JSON
            // YourJsonParser.parse(jsonData);
        });
    }

    // 实现其他EventHandler方法（onOpen、onError等）
    @Override
    public void onOpen() {}

    @Override
    public void onError(Throwable t) {}
}

在MainActivity中初始化并使用：

private void newThingCreated() throws MalformedURLException {
    // 获取主线程Handler
    Handler mainHandler = new Handler(Looper.getMainLooper());
    BackgroundEventHandler tcHandler = new BackgroundEventHandler(mainHandler);

    BackgroundEventSource bes = new BackgroundEventSource.Builder(tcHandler,
            new EventSource.Builder(
                    ConnectStrategy.http(explorer.createQueryUrl("/events/thing_created"))
                            .connectTimeout(10, TimeUnit.SECONDS)
            )
    )
            .threadPriority(Thread.MAX_PRIORITY)
            .build();
    bes.start();
    // 移除原有的messages().forEach代码
}

方式2：使用Activity的runOnUiThread（需避免内存泄漏） #

通过弱引用持有Activity，在onMessage中调用runOnUiThread：

public class BackgroundEventHandler implements EventHandler {
    private static final String TAG = "BackgroundEventHandler";
    private final WeakReference<MainActivity> activityRef;

    public BackgroundEventHandler(MainActivity activity) {
        this.activityRef = new WeakReference<>(activity);
    }

    @Override
    public void onMessage(String event, MessageEvent messageEvent) {
        Log.d(TAG, messageEvent.getData().toString());
        MainActivity activity = activityRef.get();
        // 检查Activity是否存活，避免内存泄漏和空指针
        if (activity != null && !activity.isFinishing()) {
            activity.runOnUiThread(() -> {
                // 主线程处理逻辑
                String jsonData = messageEvent.getData();
                // 处理JSON、更新UI等操作
            });
        }
    }

    // 实现其他EventHandler方法
    @Override
    public void onOpen() {}

    @Override
    public void onError(Throwable t) {}
}

在MainActivity中使用：

private void newThingCreated() throws MalformedURLException {
    BackgroundEventHandler tcHandler = new BackgroundEventHandler(this);

    BackgroundEventSource bes = new BackgroundEventSource.Builder(tcHandler,
            new EventSource.Builder(
                    ConnectStrategy.http(explorer.createQueryUrl("/events/thing_created"))
                            .connectTimeout(10, TimeUnit.SECONDS)
            )
    )
            .threadPriority(Thread.MAX_PRIORITY)
            .build();
    bes.start();
    // 移除原有的messages().forEach代码
}

原理说明 #

BackgroundEventSource会在后台线程执行EventHandler的所有回调方法，因此onMessage本身运行在后台线程。我们通过Handler.post()或runOnUiThread()将消息处理逻辑投递到主线程的消息队列，既遵守了Android的线程模型，又能安全地在主线程处理SSE消息。

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