可观察到按需发射物品

Question

我想创建一个可观察的，将按需发射项目，这意味着我想要一个单一的订阅可观察的，并通知可观察的，我需要基于我的请求的新项目。

这就是我使用Publishsubject所做的：

public class RecognizeSubject {

PublishSubject<Bitmap> mSubject;

private Context mContext;
private FaceDetector mFaceDetecor;

public RecognizeSubject(Context mContext) {
    this.mContext = mContext;
    this.mSubject = PublishSubject.create();
}

public void detect(Bitmap btm){
    mSubject.onNext(btm);
}

public Flowable<SinglePhotoId> execute() {
    return mSubject.toFlowable(BackpressureStrategy.DROP)
            .observeOn(Schedulers.newThread())
            .map(bitmap1 -> recognize(bitmap1))
            .observeOn(AndroidSchedulers.mainThread())
            .doOnSubscribe(disposable -> initialize())
            .doFinally(() -> release());
}


private void initialize() {
    mFaceDetecor = new FaceDetector.Builder(mContext)
            .setTrackingEnabled(false)
            .setLandmarkType(FaceDetector.ALL_LANDMARKS)
            .build();
}

private void release() {
    if (mFaceDetecor != null)
        mFaceDetecor.release();
}

private SinglePhotoId recognize(Bitmap bitmap) {
    //SystemClock.sleep(3000);
   //make hard background work and return SinglePhotoId object
}

}

下面是活动类中的用法：

private void takeSubjectSnap() {
    if (mSubject == null)
        mSubject = new RecognizeSubject(getBaseContext());

    if (mDisposable == null || mDisposable.isDisposed()) {
        mDisposable = mSubject.execute()
                .subscribe(this::handleDetectionSuccess,
                        this::handleDetectionError,
                        this::handleDetectionCompleted);
    }

    mSnapshotButton.setProgress(true);
    mSubject.detect(myVideoView.getBitmap());
}

所以基本上我订阅了Flowable对象，并将Bitmap对象传递给我的Subject类，通过Flowable继续并返回结果，这个解决方案是正确的还是会产生一些内存泄漏？

是否有更好的解决方案将对象发送到Observable，以便通过标准onNext（）方法继续并返回结果？

Answer 1

好的，我已经阅读了你的答案，并对我的代码做了一些修改，下面是一个图表，说明了我想要实现的目标：

因此，我的Subject类将在后台线程上处理接收到的数据，并通过onNext（）方法将处理过的项目发送给观察者。我制作了一个简单的主题，它接收Integer对象并将其转换为String对象，下面是代码：

public class MySubject {

private String TAG = "MySubject";

private PublishSubject<Integer> subject;
private final Observable<String> observable;

public MySubject() {
   Log.d(TAG, "---> MySubject() called");
   this.subject = PublishSubject.create();
   this.observable = subject
           .doOnSubscribe(disposable -> init())
           .doFinally(() -> relese()) //try do after terminate
           .observeOn(Schedulers.newThread())
           .map(this::myMap)
           .observeOn(AndroidSchedulers.mainThread());
}


private void init(){
   Log.d(TAG, "---> init() called");
}

private void relese(){
   Log.d(TAG, "---> relese() called");
}

private String myMap(Integer integer){
   Log.d(TAG, "---> myMap() called int: " + integer);
   SystemClock.sleep(3000);
   return " :) " + String.valueOf(integer);
}

public void decode(Integer integer){
   subject.onNext(integer);
}

public Observable<String> getObservable(){
   return observable;
}

}

下面是活动类中的一个用法：

Disposable disposable = Disposables.disposed();
MySubject subject = new MySubject();

void onButton1() {

   if(disposable.isDisposed()){
       disposable = subject.getObservable()
               .subscribe(s -> {
                   Log.d(TAG, "---> onNext() called " + s);
               }, throwable -> {
                   Log.d(TAG, "---> onError() called " + throwable.getMessage());
               }, () -> {
                   Log.d(TAG, "---> onCompleted() called ");
               });
   }

   Random generator = new Random();
   int i = generator.nextInt(100) + 1;
   subject.decode(i);
}

每次调用onButton1（）方法时，我都会向主题对象发布新的随机int，然后在完成后通过onNext（）方法接收处理过的数据。

该解决方案是否正确，并且不会导致任何副作用或内存泄漏？当然，我在onStop（）活动方法中取消了subject的订阅。也许有更好的解决方案来处理Rxjava中的此类问题？

如有任何进一步的答复，将不胜感激：）

Answer 2

（我将此作为评论，因为这不是答案，但太长了）

我真的不确定您在这里的用例是什么，也不确定您所描述和实现的东西到底想要实现什么。

您所描述的是一种机制，它能够处理某些内容，并在空闲时请求新内容/能够处理/等等

您实现的是一种基于推送的处理器根据从客户端接收的数据处理项目的机制。

因此，如果您实现的东西按照您的要求工作，这很好，但我建议做一些小的更改：

• 我将execute方法重命名为其他方法（因为它不执行任何操作）
• 我想用一次性用品启动一次性。disposed（）以避免空检查和
• 我想将RecognizeSubject重命名为其他名称，因为现在它正在泄漏有关其内部实现的信息
• 我会把mSubject变成private和变成final
• 我要去掉匈牙利符号

我不确定在使用位图时，flowable是否合适，你确定你一次需要那么多位图并处理所有位图（并删除未处理的位图吗？）。

Answer 3

RX继电器在这种情况下非常方便

1. 创建一个PublishRelay主题
2. 订阅它
3. 并使用publishRelaySubject传递数据。调用（你的对象）

https://github.com/JakeWharton/RxRelay