ExtJS7自定义事件能否通过WebFlux接收Server-Side-Events？

刚好做过类似的场景，结合ExtJS现代工具包和WebFlux的Server-Sent Events（SSE）来实现你的需求完全没问题，而且完美契合你不想用轮询或WebSocket的诉求。下面给你一步步拆解实现方案：

SSE是基于HTTP的单向推送协议，正好适合你这种低频操作（CSV上传不频繁）的状态更新场景：后端WebFlux生成实时状态流，前端用原生EventSource连接，再把SSE消息转换成ExtJS的自定义事件，让UI组件能轻松响应状态变化——完全不需要持续轮询，也比WebSocket轻量得多。

首先在Spring Boot里写一个返回SSE流的接口，用来推送CSV处理的进度和状态：

@RestController
@RequestMapping("/api/csv")
public class CsvProcessingController {

    @Autowired
    private CsvProcessingService processingService;

    // 返回SSE流，媒体类型指定为TEXT_EVENT_STREAM_VALUE
    @GetMapping(value = "/process/{taskId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<CsvProcessingStatus>> streamProcessingStatus(@PathVariable String taskId) {
        return processingService.getProcessingStatusStream(taskId)
                .map(status -> ServerSentEvent.<CsvProcessingStatus>builder()
                        .id(String.valueOf(status.getProgress())) // 消息ID，可选
                        .event("csv-progress") // 自定义事件名，前端会监听这个事件
                        .data(status) // 推送的状态数据（比如进度百分比、当前状态描述）
                        .build());
    }
}

对应的CsvProcessingService需要维护一个实时状态流，比如用ReplaySubject或者FluxSink来在CSV解析的每一步推送更新：

@Service
public class CsvProcessingService {
    // 用Map存储每个任务的状态流，支持多用户并行处理
    private final Map<String, FluxSink<CsvProcessingStatus>> statusSinks = new ConcurrentHashMap<>();

    // 启动CSV处理的方法，返回任务ID
    public String startProcessing(MultipartFile file) {
        String taskId = UUID.randomUUID().toString();
        // 创建状态流
        Flux<CsvProcessingStatus> statusFlux = Flux.create(sink -> {
            statusSinks.put(taskId, sink);
            sink.onDispose(() -> statusSinks.remove(taskId));
        });

        // 异步处理CSV，逐步更新状态
        CompletableFuture.runAsync(() -> {
            try {
                // 模拟解析步骤，实际替换成你的CSV解析逻辑
                for (int i = 0; i <= 100; i += 10) {
                    Thread.sleep(500); // 模拟处理耗时
                    sendStatusUpdate(taskId, i, "正在解析第" + (i/10) + "段数据");
                }
                sendStatusUpdate(taskId, 100, "处理完成");
            } catch (Exception e) {
                sendStatusUpdate(taskId, -1, "处理失败：" + e.getMessage());
            }
        });

        return taskId;
    }

    // 推送状态更新
    private void sendStatusUpdate(String taskId, int progress, String statusDesc) {
        FluxSink<CsvProcessingStatus> sink = statusSinks.get(taskId);
        if (sink != null && !sink.isCancelled()) {
            sink.next(new CsvProcessingStatus(progress, statusDesc));
            if (progress == 100 || progress == -1) {
                sink.complete();
            }
        }
    }

    // 获取状态流供Controller返回
    public Flux<CsvProcessingStatus> getProcessingStatusStream(String taskId) {
        return Flux.create(sink -> {
            FluxSink<CsvProcessingStatus> existingSink = statusSinks.get(taskId);
            if (existingSink != null) {
                // 绑定已有流到当前请求的sink
                existingSink.onNext(sink::next);
                existingSink.onComplete(sink::complete);
                existingSink.onError(sink::error);
            } else {
                sink.error(new RuntimeException("任务不存在"));
            }
        });
    }

    // 状态实体类
    public static class CsvProcessingStatus {
        private int progress;
        private String status;

        // 构造器、getter、setter省略
    }
}

ExtJS现代工具包不需要依赖旧插件，直接用原生EventSource结合ExtJS的事件系统即可，推荐把SSE逻辑封装成可复用的单例，再通过自定义事件分发状态：

2.1 封装SSE管理器（单例） #

创建一个全局的SSE管理类，统一处理连接、断开和消息分发：

Ext.define('MyApp.util.SseManager', {
    singleton: true,
    currentSource: null,

    // 启动SSE监听，传入任务ID
    startListening: function(taskId) {
        // 先关闭之前的连接，避免资源泄漏
        this.stopListening();

        const url = `/api/csv/process/${taskId}`;
        this.currentSource = new EventSource(url);

        // 监听后端定义的"csv-progress"事件
        this.currentSource.addEventListener('csv-progress', (event) => {
            const status = Ext.decode(event.data);
            // 触发ExtJS全局自定义事件，让所有组件都能监听
            Ext.GlobalEvents.fireEvent('csvprocessingupdate', status);
        });

        // 处理连接错误
        this.currentSource.onerror = (error) => {
            console.error('SSE连接异常:', error);
            this.stopListening();
            Ext.GlobalEvents.fireEvent('csvprocessingerror', error);
        };
    },

    // 停止监听并关闭连接
    stopListening: function() {
        if (this.currentSource) {
            this.currentSource.close();
            this.currentSource = null;
        }
    }
});

2.2 在UI组件中监听自定义事件 #

在你的上传面板里，用户点击处理按钮后先上传文件获取任务ID，再启动SSE监听，然后通过全局事件更新UI：

Ext.define('MyApp.view.CsvUploadPanel', {
    extend: 'Ext.panel.Panel',
    xtype: 'csvuploadpanel',

    items: [
        {
            xtype: 'filefield',
            fieldLabel: '选择CSV文件',
            name: 'csvFile',
            accept: '.csv'
        },
        {
            xtype: 'button',
            text: '开始处理',
            margin: '10 0',
            handler: function(btn) {
                const panel = btn.up('csvuploadpanel');
                const fileField = panel.down('filefield');
                const file = fileField.fileInputEl.dom.files[0];

                if (!file) {
                    Ext.Msg.alert('提示', '请先选择CSV文件');
                    return;
                }

                // 第一步：上传CSV文件，获取任务ID
                Ext.Ajax.request({
                    url: '/api/csv/upload',
                    method: 'POST',
                    headers: {
                        'Content-Type': 'multipart/form-data'
                    },
                    form: fileField.el.dom.up('form'),
                    success: function(response) {
                        const result = Ext.decode(response.responseText);
                        const taskId = result.taskId;

                        // 第二步：启动SSE监听状态
                        MyApp.util.SseManager.startListening(taskId);

                        // 显示进度条
                        panel.down('progressbar').show();
                    },
                    failure: function() {
                        Ext.Msg.alert('错误', '文件上传失败');
                    }
                });
            }
        },
        {
            xtype: 'progressbar',
            hidden: true,
            text: '初始化处理...'
        }
    ],

    initComponent: function() {
        this.callParent(arguments);

        // 监听全局自定义事件，更新UI
        Ext.GlobalEvents.on('csvprocessingupdate', this.updateProcessingStatus, this);
        Ext.GlobalEvents.on('csvprocessingerror', this.handleProcessingError, this);
    },

    // 更新处理状态和进度条
    updateProcessingStatus: function(status) {
        const progressBar = this.down('progressbar');
        if (status.progress === -1) {
            progressBar.updateText(`处理失败：${status.status}`);
            MyApp.util.SseManager.stopListening();
            return;
        }

        progressBar.updateProgress(status.progress / 100, `${status.status} (${status.progress}%)`);

        // 处理完成后停止监听并提示
        if (status.progress === 100) {
            MyApp.util.SseManager.stopListening();
            Ext.Msg.alert('完成', 'CSV文件处理完成！');
            progressBar.hide();
        }
    },

    // 处理错误
    handleProcessingError: function(error) {
        MyApp.util.SseManager.stopListening();
        Ext.Msg.alert('错误', '处理过程中出现异常，请重试');
        this.down('progressbar').hide();
    },

    // 组件销毁时清理监听
    destroy: function() {
        MyApp.util.SseManager.stopListening();
        Ext.GlobalEvents.un('csvprocessingupdate', this.updateProcessingStatus, this);
        Ext.GlobalEvents.un('csvprocessingerror', this.handleProcessingError, this);
        this.callParent(arguments);
    }
});

• 任务ID唯一性：后端用UUID生成唯一任务ID，确保多用户并行处理时状态流不冲突。
• 连接生命周期：处理完成、组件销毁或出错时，一定要关闭EventSource连接，避免浏览器资源泄漏。
• 跨域配置：如果前后端跨域，后端需要配置CORS允许text/event-stream类型的请求，并且允许EventSource的相关头（比如Access-Control-Allow-Origin）。
• ExtJS事件作用域：用Ext.GlobalEvents分发事件是最灵活的方式，适合多个组件需要监听状态的场景；如果只有单个组件需要，也可以在组件内部直接处理SSE消息。

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