title: Rxjava, SubjectSubscriptionManager has too many subjectObserver
body: <p>I use a PublishSubject to make an RXBus.
  When I send a message, call <strong>subject.onNext</strong>
  And I use <strong>subject.ofType.doOnNext</strong> to register it.</p>
  
  <p>Then I found when I register one and send messages for thousands times.
  It becomes very slow. I add some log,and I know the SubjectObserver is too many.
  The loop in PublishSubject.onNext cost lots of time.</p>
  
  <p>Then I add a method to clear some used subject.</p>
  
  <pre><code>        public synchronized State removeFinishObserver(){
              SubjectObserver[] a = observers;
              SubjectObserver[] b = new SubjectObserver[a.length + 1];
              int index = 0;
              for(SubjectObserver observer:observers){
                  LogUtils.d("onNextFinish:" + observer.onNextFinish);
                  if(!observer.onNextFinish){
                      b[index] = observer;
                      index ++;
                  }
              }
              SubjectObserver[] c = new SubjectObserver[index];
              return new State(terminated, c);
      }
  </code></pre>
  
  <p>But it doesn't work well,and even slower.I don't have idea to make it quickly.So I doubt anyone can help me.</p>
  
  <p>======================some more======================</p>
  
  <pre><code>void sendNext();//after several millseconds, receive will happened,maybe timeout
  void receive(){
      RxBus.getDefault().post(tmp);
  }
  RxBus.getDefault().toObservable(String.class)
              .doOnNext(s -&gt; {
                      sendNext();
                  }
              })
              .timeout(5, TimeUnit.SECONDS)
              .doOnError(throwable -&gt; {
                  if(throwable instanceof TimeoutException){
                      sendNext();
                  }
              })
              .retry();
  public class RxBus {
      private static volatile RxBus defaultInstance;
  
      private final Subject&lt;Object, Object&gt; bus;
      public RxBus() {
          bus = new SerializedSubject&lt;&gt;(PublishSubject.create());
      }
  
      public static RxBus getDefault() {
          if (defaultInstance == null) {
              synchronized (RxBus.class) {
                  if (defaultInstance == null) {
                      defaultInstance = new RxBus();
                  }
              }
          }
          return defaultInstance ;
      }
      public void post (Object o) {
          bus.onNext(o);
      }
      public &lt;T&gt; Observable&lt;T&gt; toObservable (Class&lt;T&gt; eventType) {
          return bus.ofType(eventType);
      }
  }
  
  compile 'io.reactivex:rxandroid:1.2.1'
  compile 'io.reactivex:rxjava:1.2.3'
  </code></pre>