开发者社区

文档建议反馈控制台

最新优惠活动

文章/答案/技术大牛

发布

为什么要在`single`之前调用`toBlocking`？

在RxJava中，toBlocking是一个用于将Observable转换为阻塞式的Observable的操作符。而single是一个用于获取Observable发射的单个数据项的操作符。为了保证能够在调用single之前获取到Observable发射的数据项，需要先调用toBlocking将Observable转换为阻塞式的Observable。

使用toBlocking的主要目的是为了在当前线程中阻塞等待Observable发射数据，并返回数据项，从而可以在后续的代码中直接使用这个数据项。如果不调用toBlocking，而是直接调用single，由于Observable是异步发射数据的，可能导致在single被调用时，Observable还未发射任何数据，从而无法获取到想要的数据项。

调用toBlocking将Observable转换为阻塞式的Observable后，可以使用single操作符获取Observable发射的单个数据项。single操作符有以下特点：

如果Observable发射了多个数据项，或者没有发射任何数据项，则会抛出IllegalArgumentException异常。
如果Observable正常终止（发射了一个数据项或者完成），则会返回该数据项。
如果Observable发生了错误，则会将错误传递给调用者。

使用场景：在某些情况下，我们只关心Observable发射的单个数据项，而不关心多个数据项或者错误。此时可以使用toBlocking().single()来获取单个数据项，并在代码中进行相应的处理。例如，在一个网络请求中，我们可能只关心请求的结果数据，而不关心其它响应数据。

腾讯云相关产品和产品介绍链接地址：

腾讯云基于云原生构建的容器服务产品：云原生应用托管服务 TKE，详情请参考腾讯云容器服务TKE
腾讯云音视频处理服务：云点播 VOD，详情请参考腾讯云云点播VOD
腾讯云人工智能服务：腾讯云人工智能，详情请参考腾讯云人工智能
腾讯云物联网平台：物联网套件 IoT Explorer，详情请参考腾讯云物联网套件IoT Explorer
腾讯云移动开发平台：移动开发平台 MDP，详情请参考腾讯云移动开发平台MDP
腾讯云分布式数据库：TDSQL for MySQL，详情请参考腾讯云TDSQL for MySQL
腾讯云区块链服务：腾讯云区块链服务 TBaaS，详情请参考腾讯云TBaaS
腾讯云元宇宙平台：元宇宙 MR，详情请参考腾讯云元宇宙MR

相关搜索:为什么需要在 fetchAll() 之前执行execute()为什么我要在使用angular之前删除turbolink 为什么Android Studio要在构建之前进行清理？为什么我要在赋值之前引用局部变量？它总是需要在Keras之前安装TensorFlow吗？为什么？为什么ReaderWriterLockSlim要在EnterReadLock()中调用睡眠()呢？为什么我需要在调用函数时添加'&‘为什么在打印函数之前调用函数？为什么要在使用日志框架进行日志记录之前检查loglevel？当input()用作函数参数时，为什么要在main()之前运行为什么PHP重定向要在实际重定向之前加载页面？为什么要在每次调用时往返到自省点？为什么我需要在syscall读取之前调用syscall write才能让它接受用户输入？为什么我要在HttpPut调用中使用路由和正文？为什么我的函数在storage.single中被调用了两次？为什么我们需要在redux thunk或saga中调用api，为什么不在组件中调用？Postgres测试容器-如果存在级联，为什么要在删除表之前删除约束？为什么要在更改react状态之前或在处理它时克隆react状态？(DRF)当创建unique=True时，为什么要在创建之前进行查询？为什么在解析服务调用之前呈现模板

相关搜索:

页面内容是否对你有帮助？

有帮助

没帮助

相关·内容

为什么wait和notify方法要在同步块中调用？

; synchronized (lock) { lock.wait(); lock.notifyAll(); } } 也就是说wait和notify方法要在基于同一个对象的同步块中使用...，那么这是为什么呢？...为什么wait和notify方法要在同步块中调用？我们先来发出一个灵魂拷问：什么时候才需要wait? 什么时候又需要notify?...take，发现buffer.isEmpty 在消费者调用wait之前，由于cpu的调度，消费者线程被挂起，生产者调用add，然后notify 然后消费者调用wait (注意，由于错误的条件判断，导致wait...所以：wait和notify方法要在同步块中调用的根本原因是，这两个方法存在竞态条件。如果不加锁的话，那么wait被调用的时候可能wait的条件已经不满足了(如上述)。

9822 0

Hystrix线程池机制的资源隔离在业务中的最佳实践

若把对某个依赖服务的所有调用请求，全部隔离在同一份资源池内。资源隔离对这个依赖服务的所有调用请求，全部走这个资源池内的资源，不会去用其他的资源。...4.1 同步 new CommandHelloWorld("World").execute()， new ObservableCommandHelloWorld("World").toBlocking...("World").queue()， new ObservableCommandHelloWorld("World").toBlocking().toFuture() 对command调用queue()..., fWorld.toBlocking().single()); fWorld.subscribe(new Observer() { @Override public..., fWorld.toBlocking().single()); fWorld.subscribe(new Observer() { @Override public

3102 0

SpringCloudRPC调用核心原理：RxJava响应式编程框架，聚合操作符

String[] items = {"one", "two", "three", "four"}; Integer count = Observable .from(items) .count() .toBlocking...在上面的代码中，为了获取count输出流中的数据项，使用了toBlocking()和single()两个操作符。...其中，Observable.toBlocking()操作返回了一个BlockingObservable阻塞型实例，该类型不是一种新的数据流，仅仅是对源Observable的包装，只是该类型会阻塞当前线程...BlockingObservable.single()方法表示阻塞当前线程，直到从封装的源Observable获取到唯一的弹射数据元素项，如果Observable源流弹射出的数据元素不止一个，single...本文给大家讲解的内容是SpringCloudRPC远程调用核心原理：RxJava响应式编程框架，聚合操作符下篇文章给大家讲解的是SpringCloudRPC远程调用核心原理：RxJava响应式编程框架

4254 0

为了程序的健壮性，我们可以使用空对象模式

return new Optional(Observable.just(value)); } } public T get() { return obs.toBlocking...().single(); } public T orElse(T defaultValue) { return obs.defaultIfEmpty(defaultValue...).toBlocking().single(); } } 上面讲了那么多，是为了防止空指针出现，现在我们来看看空对象模式具体的使用场景吧，假设我们在代码中使用了链式调用，形如： client =...理论上，第一次调用createConnection()方法时，connection是不会为空的。但是在使用某个APM sdk时，确实发现有极少的概率connection会为空。...链式调用，会让代码更加清晰，带给我们的好处是非常明显的。只要我们处理好空指针的隐患，就可以更愉悦地写代码了O(∩_∩)O哈哈~

5662 1

RxJava1 升级到 RxJava2 所踩过的坑

ObservableEmitter 可以理解为发射器，是用来发出事件的，它可以发出三种类型的事件，通过调用emitter的onNext(T value)、onComplete()和onError(Throwable...需要特别注意，emitter的onComplete()调用后，Consumer不再接收任何next事件。 6....DataUtils.isAvailable(data); } }).firstElement().toObservable().publish(); 9. toBlocking...return new Optional(Observable.just(value)); } } public T get() { return obs.toBlocking...).toBlocking().single(); } } 升级到RxJava2之后，get() 和 orElse() 方法都会报错，修改之后是这样的。

1.4K3 0

Hystrix熔断和限流(一)

; public Future queue() { final Observable o = toObservable(); final Future f = o.toBlocking...Observable fWorld = new CommandHelloWorld("World").observe(); String result = fWorld.toBlocking...().single(); 通过Observable对象, 还可以进行业务扩展 onNext(): 是run()逻辑执行完毕后的自定义逻辑, 参数为run()返回值. onCompleted(): run...Command逻辑执行失败时, 会调用getFallback()方法, 执行熔断逻辑. public class CommandHelloFailure extends HystrixCommand<String

7121 0

每日一水rx-java

数据消息流的人与元素打包成新的object输出 * 一对多 * 可以转换同样可以改变弹射的数据类型 * 多个object合并 * scan操作符累积然后弹射，中间结果 5 聚合型操作符 * count数据项统计 * toBlocking...阻塞实例 * single阻塞线程 * reduce 规约函数最后结果 6 其他操作符 * take 数据挑选n个元素，skip是跳过 * window弹射固定窗口的主题，支持滑动窗口 7 RxJava...的Scheduler调度器 Scheduler调用 * Scheduler.io()获取内部的ioScheduler。...当前线程有运行则等待 * Scheduler.single使用内置的单线程执行Rxjava流操作。

3190 0

Feign整合Ribbon和Hystrix源码解析

isReturnsObservable(method)) { // Create a cold Observable return ((Observable) result).toBlocking...first(); } else if (isReturnsSingle(method)) { // Create a cold Observable as a Single...return ((Single) result).toObservable().toBlocking().first(); } else if (isReturnsCompletable...hystrixCommand.toObservable(); } else if (isReturnsSingle(method)) { // Create a cold Observable as a Single...shouldClose) { ensureClosed(response.body()); } } } 再往下深入就是Ribbon的负载均衡了，具体内容可以参考之前的文章

9851 0

Spring Cloud Feign - 内部实现细节

概述 Feign用于服务间调用，它的内部实现是一个包含Ribbon(负载均衡)的JDK-HttpURLConnection(Http)调用。...虽然调用形式是类似于PRC，但是实际调用是Http，这也是为什么Feign被称为伪RPC调用的原因。内部调用过程如下: ? 2....return Observable.error(e); } } }) .toBlocking...() .single(); } catch (Exception e) { Throwable t = e.getCause(); if...headers) .request(request) .body(stream, length) .build(); } 参考 Spring Cloud Feign优化为什么要用

6741 0

Spring Cloud Feign - 内部实现细节

概述 Feign用于服务间调用，它的内部实现是一个包含Ribbon(负载均衡)的JDK-HttpURLConnection(Http)调用。...虽然调用形式是类似于RPC，但是实际调用是Http，这也是为什么Feign被称为伪RPC调用的原因。内部调用过程如下: 2....return Observable.error(e); } } }) .toBlocking...() .single(); } catch (Exception e) { Throwable t = e.getCause(); if...headers) .request(request) .body(stream, length) .build(); } 参考 Spring Cloud Feign优化为什么要用

3096 0

三十六、Hystrix请求命令：HystrixCommand和HystrixObservableCommand

// Observable observe = command.toObservable(); System.out.println("==========订阅者之前的一句话...} } 运行程序，控制台打印： // 使用observe()方法结果：发射数据动作立马执行开始发射数据：1 开始发射数据：2 开始发射数据：3 开始发射数据：4 ==========订阅者之前的一句话...====== 1 2 3 4 ==========订阅者之后的一句话========== // 使用toObservable()方法结果：发射数据动作明显是木有立马开始的 ==========订阅者之前的一句话...新订阅者连历史数据也能够监听到（1分钟内） queue()：调用toObservable().toBlocking().toFuture()返回 Future 对象 execute()：调用#queue...execute()、queue()也都使用了RxJava来实现，并且queue()是通过toObservable()来获得一个Cold Observable（不会立马执行），并且通过toBlocking

3.5K2 0

SpringCloud - Hystrix的执行流程

HystrixCommand仅返回一个结果的调用。 HystrixObservableCommand可能会返回多条结果的调用。直接继承HystrixCommand并实现run方法即可。...command的执行方法执行Command就可以发起一次对依赖服务的调用要执行Command，需要在4个方法中选择其中的一个前两种是HystrixCommand独有的哦 2.1 execute...queue() { /* * 当Future.cancel（boolean）的“ mayInterrupt”标志设为true时 * 由Observable.toBlocking...* 因此，为了遵守Future的约定，我们必须围绕它. */ final Future delegate = toObservable().toBlocking...queue().get() 在 queue() 方法中,会调用toObservable().toBlocking().toFuture() 即,无论是哪种执行command的方式，最终都是依赖

3621 0

【一起学源码-微服务】Hystrix 源码三：Hystrix核心流程：Hystix降级、熔断等原理剖析

OPEN/HALF_OPEN) 这里我们以AbstractCommand.applyHystrixSemantics() 为入口，一步步往下探究，这个方法在上一讲已经提到过，一个正常的Feign请求都会调用此方法...此时会进入到之前的handleFallback回调函数中，这里又会更新HealthCounts中的数据，对应的触发之前熔断的判断条件： protected HystrixCircuitBreakerImpl...isReturnsObservable(method)) { // Create a cold Observable return ((Observable) result).toBlocking...().first(); } else if (isReturnsSingle(method)) { // Create a cold Observable as a Single...return ((Single) result).toObservable().toBlocking().first(); } else if (isReturnsCompletable

1.2K1 0

【一起学源码-微服务】Feign 源码三：Feign结合Ribbon实现负载均衡的原理分析

Response response; long start = System.nanoTime(); try { // 这个client就是之前构建的...我们可以接着看client端的调用： public class LoadBalancerFeignClient implements Client { @Override public...Observable.error(e); } } }) .toBlocking...() .single(); } catch (Exception e) { Throwable t = e.getCause();...// 省略代码 } } } 上面代码已经很清晰了，这里就是真正的通过ribbon的 rule.chooseServer()负载均衡地选择了一个服务节点调用

8171 0

深入理解Feign之源码解析

Feign的源码地址：https://github.com/OpenFeign/feign 写一个Feign 在我之前的博文有写到如何用Feign去消费服务，文章地址：http://blog.csdn.net...其中value()和name()一样，是被调用的 service的名称。...//省略代码 } 从代码@ConditionalOnClass(ApacheHttpClient.class)注解可知道，只需要在pom文件加上HttpClient的classpath就行了，另外需要在配置文件上加上...Observable.error(e); } } }) .toBlocking...() .single(); } catch (Exception e) { Throwable t = e.getCause();

4.9K8 2

Spring Cloud Ribbon 全解（3） - 基本组件实现源码（1）

所有Ribbon负载均衡器需要实现的接口IClient 对于这个IClient，之前我们说到执行器逻辑，例如重试还有异常处理，都在这里处理。...return Observable.error(e); } } }) .toBlocking...() .single(); } catch (Exception e) { Throwable t = e.getCause(); if.../最多尝试几个服务实例 final int maxRetrysNext = retryHandler.getMaxRetriesOnNextServer(); //对于每个服务实例的调用逻辑...ServerStats stats = loadBalancerContext.getServerStats(server); //每个Server包含重试逻辑的请求调用

5771 0

hystrix工作原理及源码解析

Future中异步地获得响应的结果： public Future queue() { /* * The Future returned by Observable.toBlocking...executionHook.onExecutionStart(_cmd); //在这里会调用...//其中该run()为自定义的command实现的run方法，即业务执行部分的代码 //看代码之前需要熟悉下...buckets .flatMap(reduceWindowToSummary) //convert a window of bucket-summaries into a single...some amount of time (HystrixCommandProperties.circuitBreakerSleepWindowInMilliseconds()), the next single

1.6K4 0

六十、Ribbon具有负载均衡能力的客户端：AbstractLoadBalancerAwareClient

AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig)); }) // 阻塞的，顺序执行使用RxJava是为了编程方便、优美 .toBlocking...() .single(); } } 说明：为了代码结构清晰，内嵌的很多try…catch均省略了该方法的核心要义是：使用LoadBalancerCommand包装execute目标方法...因此在实际应用中：请勿直接调用execute方法，而是使用更加上层、功能更强的executeWithLoadBalancer()方法。...IClientConfig接口的自带实现仅有DefaultClientConfigImpl // 注意config.loadProperties(name)方法会被调用哦（配置会被加载进来） //...> client = null; ILoadBalancer loadBalancer = null; // 如果同名的Client已经创建过了，在调用此方法就抛错，而并非把缓存里的返回给你

1.6K2 0

再谈openfeign,聊聊它的源代码

LoadBalancerFeignClient)client).getDelegate(); } builder.client(client); } //这里是HystrixTargeter，不知道为什么总是不用...的invoke，这里最终调用了底层的OkHttpClient。...return Observable.error(e); } } }) .toBlocking...() .single(); } catch (Exception e) { Throwable t = e.getCause(); if...() .single(); } catch (Exception e) { Throwable t = e.getCause(); if

7111 0

Spring Cloud Zuul重试机制探秘

注册中心(Eureka Server) 创建一个Eureka Server只需要在主函数上添加 @EnableEurekaServer，并在 properties文件进行简单配置即可，具体内容如下： @...() .single(); } catch (Exception e) { Throwable t = e.getCause(); if...我原本的想法是这个请求被包装成Observable，如果这次请求因为超时出现异常或者其他异常，这样就会触发Observable的重试机制(RxJava)，但是事实并非如此，为什么呢？...原因就是上面的那两个参数，当出现了超时异常的时候，在触发重试机制之前会调用 RequestSpecificRetryHandler的 isRetriableException()方法，该方法的作用是用来判断是否执行重试动作...这里就会有人问，因为最外层是采用Hystrix，而Hystrix此时已经超时了，为什么还允许它内部继续使用spring-retry进行重试呢？

4.3K10 0

点击加载更多

扫码

添加站长进交流群

领取专属 10元无门槛券

手把手带您无忧上云

扫码加入开发者社群

相关资讯

热门标签

活动推荐

运营活动

活动名称

广告关闭