在.subscribe()之后添加.dispose()有什么问题
在.subscribe()之后添加.dispose()会导致订阅的观察者被立即取消，不再接收任何后续的数据流。这样做可能会导致以下问题：
- 数据流不完整：如果在.dispose()之后还有其他重要的数据流需要被观察，那么这些数据将无法被订阅者接收到。
- 内存泄漏：如果.dispose()没有正确地释放相关的资源，可能会导致内存泄漏问题。观察者可能仍然保留对被观察对象的引用，而无法被垃圾回收。
正确的做法是在不再需要订阅时，使用.unsubscribe()来取消订阅，而不是直接在.subscribe()之后添加.dispose()。这样可以确保订阅者能够接收到完整的数据流，并且正确释放相关资源，避免潜在的内存泄漏问题。
关于腾讯云相关产品和产品介绍的链接地址，由于要求不能提及具体的品牌商，建议您可以参考腾讯云官方网站或文档，搜索相关的云计算、消息队列、数据流等服务来获取更多信息。
相关·内容
在讨论了何时处理CompositeDisposable之后，我想知道为什么我们不能在代码中的.subscribe()之后添加一个.dispose()。 Observable.just(0)
.doOnNext{...}
.subscribe()
.dispose() 如果代码是这样的，到底会发生什么呢？这样写可以吗？(而不是.addTo(compositeDisposable))
浏览 55提问于2019-10-09得票数 0
我用viewWillAppear写了一个订阅。
但它也会在首次启动的应用程序中运行一次。
当我推送到另一个视图控制器时，我使用dispose()。
然后我回到first viewcontroller，我在viewWillAppear中的订阅函数没有运行。
我的rx订阅有什么问题？
var listSubscribe:Disposable?
override func viewWillAppear(_ animated: Bool) {
super.viewWillAppear(animated)
listSubscribe = chatrooms.notifySubject
浏览 0提问于2017-08-23得票数 0
我有下面的RxJava 2代码(在Kotlin中)，它有一个可以观察到的
disposable = Observable.create<String>({
subscriber ->
try {
Thread.sleep(2000)
subscriber.onNext("Test")
subscriber.onComplete()
} catch (exception: Exception) {
浏览 4提问于2017-07-10得票数 2
回答已采纳
1回答
我正在为用RxJava实现Socket.io而奋斗。我有一个问题，再次订阅相同的可观察，所以我做了一个例子和更清晰的代码来理解。
以下是功能：
private void testObservable() throws InterruptedException {
io.reactivex.Observable observable = io.reactivex.Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter
浏览 1提问于2017-06-07得票数 1
回答已采纳
下面是我使用最新预发行版WampSharp的非常简单的代码：
var channelFactory = new DefaultWampChannelFactory();
var channel = channelFactory.CreateMsgpackChannel("wss://api.poloniex.com", "realm1");
await channel.Open();
var realmProxy = channel.RealmProxy;
Console.Wri
浏览 16提问于2016-07-23得票数 8
2回答
当类在RX框架中使用时，我们如何释放类中的资源？我在C# 4.0lib项目中有一个类，它包含web服务代理、ado.net对象等，我也实现了IDisposable。这个类有一个run running方法，在这个方法中，我将数据保存到dataset中，最后当该方法完成时，我将dataset保存到DB中。
在Dispose方法中，我将dataset设置为null，并将其他web服务代理对象设置为null。然而，当该类在反应式扩展方法中频繁使用时，它会抛出内存异常。
public class MyClass : IDisposable
{
proxy object;
DataS
浏览 1提问于2013-02-19得票数 1
我一直在研究一书中的示例，这本书针对的是版本1而不是2。
Observable<BigInteger> naturalNumbers = Observable.create(subscriber -> {
Runnabler = () -> {
BigInteger i = ZERO;
while (!subscriber.isUnsubscribed()) {
subscriber.onNext(i);
i = i.add(ONE);
}
};
ne
浏览 1提问于2017-06-13得票数 7
回答已采纳
1回答
我们在LoginViewModel中有以下代码
private MutableLiveData<Response> loginLiveData;
@Inject
public LoginViewModel(LoginUseCase loginUseCase) {
loginLiveData = new MutableLiveData<>();
this.loginUseCase = loginUseCase;
}
@Override
public void onAttached() {
checkHasToken();
}
public vo
浏览 1提问于2018-10-30得票数 0
回答已采纳
我使用DynamicData (6.9.1.2588)来转换项目，然后对它们进行排序。 ObservableCollectionEx.ToObservableChangeSet<ReactiveList<DataItem>, DataItem>(model.Children)
.Filter((d) => d.ViewClass == NormalizedType.TableRow)
.Transform(d => new TableRow(d, this))
浏览 15提问于2019-05-22得票数 1
2回答
在我的应用程序中，有大量的oledbConnections、数据表和数据源以及更多的非托管资源。它们中的一些被释放(当不再需要它们时调用Dispose() )，而另一些没有Dispose() (只是为了使代码行更多更少!)。问题是:如果这些非托管资源在使用后没有被处理，在applictaion上会有什么问题吗？
我有时(并非所有时间)都会出现错误"appname已停止工作..“，这个错误是由未释放的对象引起的吗？
最后，在调用Dispose()时发生错误的情况是什么？
谢谢。
浏览 3提问于2013-02-06得票数 0
回答已采纳
是否有任何理由使用startWith在主题上发出事件
let publishedSubject1 = PublishedSubject<Void>()
publishedSubject1
.startWith(Void())
.subscribe()
.dispose(by: bag)
与仅使用.onNext相比
let publishedSubject2 = PublishedSubject<Void>()
publisedSubject2
.subscribe()
.dispose(by: bag)
publishedSubj
浏览 0提问于2018-11-19得票数 0
在我使用Rx的第一步中，我被困在这里：
public class DisposableResourceDemo : IDisposable
{
public DisposableResourceDemo() {
Console.WriteLine("DisposableResourceDemo constructor.");
}
public void Dispose() {
Console.WriteLine("DisposableResourceDemo.Dispose()");
}
浏览 0提问于2012-07-18得票数 0
回答已采纳
我有以下几点：
Flux<String> flux = ...
Mono<Void> mono = ...
Mono<Void> combined = operation(flux, mono);
它们代表并行发生的操作。
现在，我想将flux发出的所有元素输出到sysout，直到mono完成，这里使用的正确运算符是什么？
我试过了：
final Disposable subscribe = flux.subscribe(System.out::println);
mono.doOnSuccessOrError((o, e) -> subscrib
浏览 1提问于2019-06-11得票数 0
1回答
从subscribe方法中处理observable有没有明确而聪明的解决方案？例如，我有一个Observable.interval，每秒都会发出很长的信号。在我的subscribe方法中，我想检查是否有20秒已经过去，而不是取消订阅。
val observable = Observable.interval(1000,TimeUnit.MILLISECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscri
浏览 0提问于2020-03-04得票数 0
我有以下代码结构
服务
public Flowable entryFlow()
{
return Flowable.fromIterable(this::getEntries)
}
消费者
void start()
{
disposable = service
.entryFlow()
.observeOn(Schedulers.computation())
.subscribeOn(Schedulers.computation())
.subscribe(
entry -> ...,
浏览 0提问于2018-02-12得票数 4
1回答
我在我的C#项目中使用了ReactiveUI和DynamicData。但是，域模型类仍然依赖于C#事件、INotifyPropertyChanged和INotifyCollectionChanged接口。
有模型类和ViewModel类：
public class Model
{
public ObservableCollection<int> Collection { get; } = new ObservableCollection<int>();
}
public class ViewModel : ReactiveObject, IDisposable
浏览 7提问于2020-02-06得票数 2
回答已采纳
1回答
在RX订阅后释放
、、、
我有一堂课，里面有很热的，可观察的。它进行一些图像转换并收集度量(例如，为了简单起见，int)
public class Provider
{
public IObservable<(int Metric, Bitmap Image)> HotObservable { get; }
public Provider()
{
HotObservable = Observable.Interval(TimeSpan.FromSeconds(5))
.Select(x => ...) // Some processing here
.Wh
浏览 3提问于2020-05-28得票数 1
回答已采纳
在RX方面需要帮助。我想要定义可观察性，它应该在第一次创建订阅时创建资源，为每个新订阅发布该资源实例一次，并且当完成所有订阅时，必须释放该资源实例。类似于Observable.Using，但是具有发布(值)和RefCount行为。我用标准运算符来表达它的所有尝试都失败了。这个代码可以做我想做的事情，但是我认为必须有标准的方法来完成它。我真的不想重新发明轮子。
using System;
using System.Reactive.Linq;
using System.Reactive.Disposables;
namespace ConsoleApplication1
{
class
浏览 2提问于2013-10-15得票数 1
回答已采纳
大多数Flowable.subscribe()重载返回一个允许清理流的Disposable。我有这样的习惯：
Disposable d = Flowable.just()
.map(...)
.subscribe(
n -> ...
t -> ...
() -> ...
);
// someone clicks "cancel" in another thread
d.dispose();
但是，在使用.subscribe(Subscriber)时，不返回Disposable。我想使用.
浏览 0提问于2020-06-24得票数 3
回答已采纳
1回答
我有课
class A {
IObservable<long> poll = new Observable.Interval(100 ms).Do((ms) => LoadData());
void Subscribe() {
poll.Subscribe();
}
}
我有课
class B {
IEnumerable<A> Items { get; }
void Refresh() {
Items = GetNewListWithNewJustCreatedInstances();
浏览 5提问于2016-08-10得票数 0
我很难理解Observable.Delay是如何工作的，以及Dispose()是何时被调用的。熟悉Rx的人能帮上忙吗？
以下代码片段：
static void Main(string[] args)
{
var oneNumberEveryFiveSeconds = new SomeObservable();
// Instant echo
oneNumberEveryFiveSeconds.SubscribeOn(Scheduler.ThreadPool).Subscribe(num => Console.WriteLin
浏览 3提问于2010-10-20得票数 2
2回答
我有一堂课来解释我的问题：
public class DataObserver: IDisposable
{
private readonly List<IDisposable> _subscriptions = new List<IDisposable>();
private readonly SomeBusinessLogicServer _server;
public DataObserver(SomeBusinessLogicServer server, IObservable<SomeData> data)
{
浏览 0提问于2014-01-14得票数 1
回答已采纳
我有一个可观测的，它使用scan运算符进行转换，所以它总是根据当前和以前的值(或初始值)发出一个值。
Subject<String> subject = PublishSubject.create();
Observable<String> observable = subject.scan("zero", (a, b) -> a + ", " + b);
然后，第一个观察者订阅，打印所有发出的值，并在一段时间后取消订阅。
Disposable first = observable.subscribe(System.out::pri
浏览 5提问于2017-10-29得票数 1
回答已采纳
我有下面这段代码 Single.just(settings.toString())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new SingleObserver<String>() {
private Disposable disposable;
@Override public void onSubscribe(Disposable d) {
this.disposable = d;
}
@Override public void onSuccess(St
浏览 3提问于2019-12-15得票数 1
在Rx.NET中是否有一个功能类似于BehaviorSubject的Subject实现，但只有在它发生更改时才会发出下一个值？ 我对反应式扩展比较陌生，我似乎找不到任何类似的东西，尽管这种模式感觉像是INotifyPropertyChanged的自然替代品。 我的简单实现是封装BehaviorSubject<T>，如下所示。与使用Observable.DistinctUntilChanged(https://docs.microsoft.com/en-us/previous-versions/dotnet/reactive-extensions/hh229508(v=vs.103
浏览 37提问于2021-09-18得票数 3
回答已采纳
我一直认为.share(replay: 1, scope: .forever)分享单一的上游订阅，不管下游用户有多少。
但是，我刚刚发现，如果下游订阅的计数降到零，它将停止“共享”，并在上游释放订阅(因为refCount()是在幕后使用的)。因此，当新的下游订阅发生时，它必须在上游重新订阅。在以下示例中：
let sut = Observable<Int>
.create { promise in
print("create")
promise.onNext(0)
return Disposables.crea
浏览 8提问于2022-05-23得票数 2
回答已采纳
3回答
我有一个类，它必须执行setup/teardown才能生成包装为可观察对象的事件。为简洁起见，我在下面的代码中使用Observable.Interval生成了一个可观察到的流。我需要在这段代码中做哪些更改才能让Start和Stop被调用两次？每次refcount递增到1时，需要调用有效的Start，每次refcount递减到0时，需要调用Stop。 static async Task Main(string[] args)
{
var provider = new SomeProvider();
var s1 = provider.
浏览 38提问于2020-07-17得票数 1
回答已采纳
有人知道合并运算符是如何实现的吗？我很惊讶地发现合并操作符可以正确地合并冷观察值：
var odd = new int[] { 1, 3 }.ToObservable().Trace("odd");
var even = new int[] { 2, 4 }.ToObservable().Trace("even");
odd.Merge(even).Dump("Merged");
输出：
odd1: Subscribe()
even1: Subscribe()
odd1: OnNext(1)
浏览 1提问于2021-04-02得票数 0
请考虑以下几点：
[Fact]
public void foo()
{
var result = new Subject<bool>();
var startCount = 0;
var completionCount = 0;
var obs = Observable
.Defer(() =>
{
++startCount;
return result.FirstAsync();
})
.Do(_ =
浏览 1提问于2016-02-29得票数 8
1回答
有一个函数从数据库查询数据并返回一个Flowable
@Dao
interface DataDao {
@Query("SELECT * FROM DataTable ORDER BY `order`")
fun listenToAll():Flowable<List<DataRoomEntity>>
}
当订阅它时，一次性服务不会被解决(见截图)。
var dataDao: DataDao
fun listenToDataDb(
浏览 11提问于2021-02-03得票数 0
我使用RxJava Flowable with Room监听表中的更改。如上所述，问题在于when the result of subscribe, a disposable, is added to a CompositeDisposable, it immediately unsubscribes。
相关代码如下：
repo.getFlowable(id)
?.doOnDispose { Timber.i("Unsubscribed") }
?.doOnDispose { Timber.i("Dispo
浏览 0提问于2019-07-11得票数 1
1回答
何时调用subscribeWith方法而不是普通订阅？用例是什么？
compositeDisposable.add(get()
.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.io())
.subscribe(this::handleResponse, this::handleError));
VS
compositeDisposable.add(get()
.observeOn(AndroidSchedulers.mainThread())
浏览 1提问于2017-06-19得票数 52
回答已采纳
我希望融合几个Android传感器的输入，并将输出公开为支持多个同时观察者的可观察(或至少可以订阅的)输出。有什么惯用的方法来处理这件事？标准库中是否有一个很好的起点？
我正在考虑将一个PublishSubject封装在一个带有委托的对象中，用于测试hasObservers以激活传感器的一个或多个subscribe方法，并将返回的Disposable封装在一个测试hasObservers以禁用传感器的代理中。类似这样的事情，尽管这已经存在一些明显的问题：
public class SensorSubject<T> {
private final PublishSubject
浏览 0提问于2018-06-24得票数 0
回答已采纳
1回答
添加脚本管理器时验证失败
、、、
我在我的主页上有几个电子邮件订阅控件。
Web可以正常工作，但不会触发验证，以防输入电子邮件地址。
<asp:UpdatePanel ID="updSubscribe" runat="server" UpdateMode="Conditional">
<ContentTemplate>
<asp:Panel ID="pnlSubscribe" runat="server" >
<asp:TextBox ID=&
浏览 3提问于2015-09-20得票数 1
我有一个游戏服务器，每个玩家都有一个计时器，就像这样：
this.player.Timer = from tick in TimerPublisher where tick % 1 == 0 select tick;
我有一些订阅的方法，比如：
this.player.Timer.Subscribe( tick => IncreseStamina() );
this.player.Timer.Subscribe( tick => IncresePower() );
//etc
所以我想要做的不是设置
IDisposable dis = //the subscribed method
浏览 1提问于2011-09-15得票数 2
这里有很多关于StackOverflow的问题，可以追溯到7年前关于使用ReactiveX和do或不使用主题的问题。这些问题中的大多数最终都会被那些在哲学基础上来回争论的人所回答，最终在实际的例子中变得相当轻，而不是“正确的”方法和“最好的”方法。
我编写了一个简单的示例类，试图将SignalR的IHubProxy.On连接到IObservable。
这是我的课，请教我如何重写这个类，这样就不需要主题了。
我只是没办法这么做。
using Microsoft.AspNet.SignalR;
using Microsoft.AspNet.SignalR.Client;
using Microso
浏览 0提问于2019-02-15得票数 0
为什么我们要使用dispose()方法？我对此有点困惑。如果我们不使用它，会发生什么问题?使用它有什么好处？
@override
void dispose() {
// TODO: implement dispose
super.dispose();
}
浏览 16提问于2020-01-02得票数 58
回答已采纳
1回答
我使用提供程序来侦听Supabase表中的实时更改。
使用MVVM通过侦听实时表"poll_options“来初始化视图。
poll_screen.dart
@override
void initState() {
asyncLoadData();
super.initState();
}
Future<void> asyncLoadData() async {
await Provider.of<PollScreenViewModel>(context, listen: false)
.subscribeToRealtimeVoteT
浏览 4提问于2021-12-21得票数 1
我有一个类，这个类有一个包含从文件读取大数据的"value“变量：
public class MyClass : IDisposable
{
public string name {get; set;}
public string value {get; set;} //I'm load huge data for this variable
public MyClass(string name, string value)
{
this.name = name;
this.value = value;
浏览 0提问于2013-04-28得票数 4
回答已采纳
1回答
我有http修补程序请求，并且有更多的令牌授权。但是Http请求从服务器获得一个未经授权的错误，当发出完全相同的请求(console.log(url+ token)，然后在Postman中从控制台复制它)时，它可以工作。
有什么问题吗？
this.getToken().subscribe((data: FormData) => {
const httpOptions = {
headers: new HttpHeaders({
'Authorization': ('Bearer ' + data['a
浏览 0提问于2019-07-19得票数 0
回答已采纳
代码(Kotlin)非常简单，但我不明白，为什么我不能使用同一个对象再次订阅？
val x = Observable.interval(1L, TimeUnit.SECONDS, Schedulers.io())
.map {
println("emitting=$it")
it.toString()
}.publish().autoConnect()
val o = object : DisposableObserver<String>() {
override fun onComplete() {}
浏览 2提问于2019-09-22得票数 0
回答已采纳
我只是想实现一个MVVMCross信使解决方案，它将使我能够在iOS应用程序或PCL发布时将信息上传到Google。
我遇到的问题是，在我发布之后，订阅delgate不会被触发。您能从静态类订阅MVVMCross信使订阅吗？
静态类中的订阅
public static class GoogleAnalyticsWrapper //: IDisposable
{
private const string TrackingId = "xxxxxxxxxxx";
private static readonly IMvxMessenger messenger;
浏览 10提问于2016-03-09得票数 0
回答已采纳
在处理最坏的情况时，我偶然发现了一个问题:在IConnectableObservable完成之前，正在处理一个有订阅的。
我写了一个人为的例子来重现这个问题。
var hotSource = Observable
.Return(1)
.Delay(TimeSpan.FromMilliseconds(500))
.Publish();
var disposable = hotSource.Connect();
这是我热辣的可观察到的。我增加了一个延迟，这样我就可以在它触发价值之前订阅它。
下面，我返回一个可观察到的，应该允许我等待直到hotSource完成。我相信我应
浏览 6提问于2016-04-08得票数 3
回答已采纳
我的代码订阅了一个事件，一旦事件被处理，它就需要取消订阅(Dispose)。不过，这看上去像是鸡蛋的问题。使用rec不起作用，我也找不到该怎么做。
有什么好的模式可以绕过这个限制吗？
let process = new Process()
let exitSubscription = process.Exited.Subscribe (
fun evntArg ->
exitSubscription.Dispose() <---------- Compiler complains here
// do more something h
浏览 5提问于2022-01-26得票数 1
回答已采纳
在阅读msdn关于反应式扩展之类的内容时，我发现了一个建议，建议我不应该实现IObservable，而应该使用Observable.Create……在我阅读本文时，我的项目已经有了一个ObservableImplementation<T>类，我将其用作IObservable源，只要我想将事件转换为可观察的事件。
我已经阅读了System.Reactive中的AbstractObservable<T>实现，我没有发现他们的代码和我的代码之间有任何主要的区别。那么，实现IObservable有什么问题呢？我可以向其中添加我自己的属性，等等...
为了完整起见，这是我的实现，
浏览 1提问于2012-05-07得票数 12
回答已采纳
我使用角cdk ovelay打开对话框。
这是代码
openDialog(component: any, dialogConfig: DialogConfig): any {
const config = new OverlayConfig();
config.height = dialogConfig.height;
config.width = dialogConfig.width;
config.positionStrategy = this.overlay.position().global().centerHorizontally().centerVe
浏览 3提问于2020-07-13得票数 2
1回答
我需要帮助。如何设置扫描超时，例如10秒，当我使用rxandroidble。
scanDisposable = rxBleClient.scanBleDevices(
new ScanSettings.Builder()
.setScanMode(ScanSettings.SCAN_MODE_LOW_LATENCY)
.setCallbackType(ScanSettings.CALLBACK_TYPE_ALL_MATCHES)
浏览 0提问于2018-04-12得票数 3
回答已采纳
在我拥有的html文件中
<p *ngIf="!checklistsready">
not redy
</p>
<p *ngIf="checklistsready">
Ready
</p>
在我的typescript文件中
checklistsready: boolean = false;
constructor(){
this.fetchChecks();
}
fetchChecks(){
this._checklistService.getAllchecks() //
浏览 15提问于2017-08-12得票数 0
回答已采纳
1回答
我有一个带有两个精灵的简单LibGdx应用程序。一种是简单的纹理，重复填充背景。这个很管用。另一种是α混合的纹理，使得角看起来比中间更暗。它被拉伸到覆盖整个屏幕。由于某种原因，这个出现在错误的位置，只是一个大的白色盒子。
这是我的代码：
public class TestGame extends ApplicationAdapter {
SpriteBatch batch;
boolean showingMenu;
Texture background;
Sprite edgeBlur;
Texture edgeBlurTex;
@Overri
浏览 2提问于2015-09-25得票数 0
回答已采纳
我需要有一个热的观察物，包装价格饲料。这是在多个区域订阅的。
observable是使用Refcount创建的，并传递给subscription.The first subscriber调用subscribe，从而启动流并获取所有事件。第二个将错过事件，直到它的订阅和后续订阅将执行相同的行为。
我的问题不是缺少事件。我希望所有的订阅者都能获得相同的数据。也就是说，流必须仅在订阅请求完成时启动。
有可能吗？
编辑:下面说明了两种方法及其问题。
1)
public void HotObservableSubscriptionWithRefCount()
{
var obs1
浏览 0提问于2018-11-16得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
