Observer

rxjava onNext 获取当前观察者id

追踪观察者的身份
下面是一个简单的例子，展示了如何在onNext中获取观察者的身份标识

import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
 
public class IdentifiableObserver<T> implements Observer<T> {
    private final String observerId;
 
    public IdentifiableObserver(String observerId) {
        this.observerId = observerId;
    }
 
    @Override
    public void onSubscribe(Disposable d) {
        // 可以在这里获取订阅关系
    }
 
    @Override
    public void onNext(T t) {
        // 在这里获取观察者的身份标识
        System.out.println("Observer ID: " + observerId);
        // 处理接收到的数据
        System.out.println("Received: " + t.toString());
    }
 
    @Override
    public void onError(Throwable e) {
        e.printStackTrace();
    }
 
    @Override
    public void onComplete() {
        // 数据处理完毕
    }
}
 
// 创建并使用观察者时指定身份标识
String observerId = "observer1";
IdentifiableObserver<String> identifiableObserver = new IdentifiableObserver<>(observerId);

// 假设有一个Observable发射数据
// Observable<String> source = ...
// source.subscribe(identifiableObserver);



Observable.just("Hello, world!")
    .doOnSubscribe(new Action0() {
        @Override
        public void call() {
            // 在这里记录或打印观察者的身份信息
            System.out.println("Observer subscribed: " + Thread.currentThread().getName());
        }
    })
    .subscribe(new Observer<String>() {
        @Override
        public void onSubscribe(Disposable d) {
            // 在这里记录或打印订阅的信息
        }
 
        @Override
        public void onNext(String s) {
            System.out.println("Received: " + s);
        }
 
        @Override
        public void onError(Throwable e) {
            e.printStackTrace();
        }
 
        @Override
        public void onComplete() {
            // 处理完成事件
        }
    });