基于WebFlux实现多应用向同一Flux推送事件的方案咨询

1. 完善Sink实例初始化 #

原MySink类未完成Sink的初始化，首先需要创建全局唯一的多播Sink实例，确保所有订阅者（浏览器客户端）能收到同一批事件：

@Component
public class MySink {
    // 初始化多播Sink，支持多订阅者+背压缓冲
    private final Sinks.Many<String> sink = Sinks.many().multicast().onBackpressureBuffer();

    public void next(String event) {
        // 处理推送结果，避免静默失败
        Sinks.EmitResult result = sink.tryEmitNext(event);
        if (result.isFailure()) {
            // 根据业务需求处理失败场景，比如日志告警
            System.err.println("事件推送失败: " + result);
        }
    }

    public Flux<String> asFlux() {
        return sink.asFlux();
    }
}

2. 暴露接收外部事件的HTTP接口 #

在UI应用中新增一个POST接口，供外部应用调用以推送事件，接口内部将事件转发到Sink：

@RestController
@RequestMapping("/external-events")
public class ExternalEventController {
    private final MySink mySink;

    // 构造注入（替代@Autowired，更符合Spring规范）
    public ExternalEventController(MySink mySink) {
        this.mySink = mySink;
    }

    // 接收外部应用的事件推送请求
    @PostMapping("/push")
    public ResponseEntity<Void> pushEvent(@RequestBody String event) {
        // 可选：添加参数校验，拒绝空事件
        if (event == null || event.isBlank()) {
            return ResponseEntity.badRequest().build();
        }
        mySink.next(event);
        return ResponseEntity.ok().build();
    }
}

3. 适配原有业务代码 #

调整TestService中的方法，使用修改后的MySink方法：

@Service
public class TestService {
    private final MySink mySink;

    public TestService(MySink mySink) {
        this.mySink = mySink;
    }

    public Flux<String> runTest1(List<Event> list1) {
        issueResponse.run(list1);
        return mySink.asFlux();
    }
}

4. 关键注意事项 #

• 安全防护：如果接口对外暴露，必须添加安全验证，比如：
  • 自定义API密钥校验：在请求头中携带密钥，接口内验证合法性
  • 集成Spring Security实现OAuth2/JWT令牌认证
• 结构化事件扩展：如果需要传递复杂数据，可将String替换为自定义DTO类，以JSON格式传输
• 跨域配置：若外部应用与UI应用跨域，需添加@CrossOrigin注解或全局CORS配置
• 容错处理：针对tryEmitNext的失败结果，可添加重试机制或告警逻辑

5. 外部应用调用示例 #

外部应用可通过HTTP POST请求推送事件，比如用curl命令：

curl -X POST -H "Content-Type: text/plain" -d "外部应用推送的事件内容" http://your-ui-app-domain:port/external-events/push

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