【Azure 事件中心】Spring Cloud Stream Event Hubs Binder 发送Event Hub消息遇见 Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially 异常

简介: 【Azure 事件中心】Spring Cloud Stream Event Hubs Binder 发送Event Hub消息遇见 Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially 异常

问题描述

开发Java Spring Cloud应用,需要发送消息到Azure Event Hub中。使用 Spring Cloud Stream Event Hubs Binder 依赖,应用执行一会就会遇见报错:reactor.core.publisher.Sinks$EmissionException: Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.

 

问题解答

从错误来看,这明显是多线程并发处理时,多个线程同时触发了onSubscribe 或 onNext 或 onError 或 onComplete 事件,而这些事件在与 Subscriber处理时只能一个一个串行处理。

因为SpringCloud的 Controller 并发请求时,会分配多个线程同时调用many.emitNext(),这时如果之前请求线程处理还未结束,新请求的线程会直接这样的报错。

异常产生的代码为:

private final Sinks.EmitFailureHandler emitFailureHandler = (signalType, emitResult) -> emitResult.equals(Sinks.EmitFailureHandler.FAIL_FAST);

Sinks 对 FAIL_FAST 和 FAIL_NON_SERIALIZED的枚举值说明参考:reactor-core/Sinks.java at main · reactor/reactor-core · GitHub

  • FAIL_FAST:表示对失败不会进行任何重试,会马上触发异常处理机制,这里就是抛出EmissionException异常。( A pre-made handler that will not instruct to retry any failure and trigger the failure handling immediately.)
  • FAIL_NON_SERIALIZED:表示会持续重试,直至成功。

如果发送到Event Hub的消息允许丢失,可以通过Try Catch捕获异常后记录日志即可。

但是,如果发送的消息不能丢失,必须成功传递到Event Hub中,就可以使用 FAIL_NON_SERIALIZED 模式。

修改为:

private final Sinks.EmitFailureHandler emitFailureHandler = (signalType, emitResult) -> emitResult.equals(Sinks.EmitResult.FAIL_NON_SERIALIZED);

 

 

参考资料

ReactorDispatcher: Making sure spec 1.3 is not violated and under race, signals are not lost upon concurrent ClosedChannelException : https://github.com/Azure/azure-sdk-for-java/issues/27320

FAIL NON SERIALIZED :

https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/core/azure-core-amqp/src/main/java/com/azure/core/amqp/implementation/handler/Handler.java#L89

https://github.com/reactor/reactor-core/blob/main/reactor-core/src/main/java/reactor/core/publisher/Sinks.java#L118

/**

* Has successfully emitted the signal

*/

OK,

/**

* Has failed to emit the signal because the sink was previously terminated successfully or with an error

*/

FAIL_TERMINATED,

/**

* Has failed to emit the signal because the sink does not have buffering capacity left

*/

FAIL_OVERFLOW,

/**

* Has failed to emit the signal because the sink was previously interrupted by its consumer

*/

FAIL_CANCELLED,

/**

* Has failed to emit the signal because the access was not serialized

*/

FAIL_NON_SERIALIZED,

/**

* Has failed to emit the signal because the sink has never been subscribed to has no capacity

* to buffer the signal.

*/

FAIL_ZERO_SUBSCRIBER;

相关文章
|
8月前
|
Java 开发工具 Spring
【Azure Application Insights】为Spring Boot应用集成Application Insight SDK
本文以Java Spring Boot项目为例,详细说明如何集成Azure Application Insights SDK以收集和展示日志。内容包括三步配置:1) 在`pom.xml`中添加依赖项`applicationinsights-runtime-attach`和`applicationinsights-core`;2) 在main函数中调用`ApplicationInsights.attach()`;3) 配置`applicationinsights.json`文件。同时提供问题排查建议及自定义日志方法示例,帮助用户顺利集成并使用Application Insights服务。
235 8
|
监控 Java C#
Spring Event 的介绍
Spring Event 是 Spring 框架中的事件驱动机制,允许组件间进行同步或异步消息传递,无需直接依赖。它包括事件(Event)、事件发布者(Publisher)和事件监听器(Listener),通过 `ApplicationEventPublisher` 广播事件,实现松耦合通信,增强模块化和可维护性。Spring 还提供了多种内置事件,如 `ContextRefreshedEvent` 和 `ContextClosedEvent`,支持同步及异步处理,并具备良好的扩展性。
302 3
|
10月前
|
消息中间件 Java Kafka
【Azure Kafka】使用Spring Cloud Stream Binder Kafka 发送并接收 Event Hub 消息及解决并发报错
reactor.core.publisher.Sinks$EmissionException: Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.
196 6
|
设计模式 Java Spring
Spring Event 的幕后
Spring Event 基于观察者模式,实现模块间松散耦合的通信。通过事件(Event)、事件发布者(Publisher)和事件监听器(Listener)三个核心组件,Spring Event 可以轻松实现业务解耦。Spring 容器在启动时会初始化 `ApplicationEventMulticaster`,扫描并注册所有事件监听器,通过调用 `multicastEvent()` 方法将事件广播给所有注册的监听器。
199 2
|
Java 应用服务中间件 nginx
【Azure Spring Apps】Spring App部署上云遇见 502 Bad Gateway nginx
【Azure Spring Apps】Spring App部署上云遇见 502 Bad Gateway nginx
169 0
|
Java Spring
【Azure Service Bus】使用Spring Cloud integration示例代码,为多个 Service Bus的连接使用 ConnectionString 方式
【Azure Service Bus】使用Spring Cloud integration示例代码,为多个 Service Bus的连接使用 ConnectionString 方式
215 0
|
安全 Java 定位技术
Android 浅度解析:AIDL & Binder (1)
Android 浅度解析:AIDL & Binder (1)
757 0
|
Java API Android开发
Android中Binder在项目中的具体使用详解
Android中Binder在项目中的具体使用详解
392 0
|
缓存 安全 Java
Android深入Binder拦截问题分析
【7月更文挑战第1天】Android Binder 拦截可实现虚拟化、测试、SDK检测、逆向分析及ROM扩展。通过Java层aidl代理,利用IBinder接口规范来拦截通信。拦截步骤包括:替换Binder服务缓存对象,如ActivityManagerService;代理ServiceManager以控制服务获取。此操作需系统权限,可能涉及安全风险和版本差异,非必要时应谨慎。
260 0
|
Java Android开发 C++
Android中的Binder概述
Android应用的开发离不开四大组件(Activity,Service,BroadcastReceiver,ContentProvider),而这四大组件所涉及的通信底层都是依赖于Binder IPC机制的。例如当进程A中的Activity要向进程B中的Service通信,这便需要依赖于Binder IPC。不仅如此,整个Android系统架构中,大量采用了Binder机制作为IPC方案,当然也存在部分其它的IPC方式,比如Zygote通信便是采用Socket。 概念:Binder是Android中的一种IPC方式,提供远程过程调用(RFC)功能。

热门文章

最新文章