简单谈谈RxJava和多线程并发

前言

相信对于RxJava,大家应该都很熟悉,他最核心的两个字就是异步,诚然,它对异步的处理非常的出色,但是异步绝对不等于并发,更不等于线程安全,如果把这几个概念搞混了,错误的使用RxJava,是会来带非常多的问题的。

RxJava与并发

首先让我们来看一段RxJava协议的原文:

Observables must issue notifications to observers serially (not in parallel). They may issue these notifications from different threads, but there must be a formal happens-before relationship between the notifications.

如上所述,RxJava对多线程并发其实并没有做非常的多保护,这段话中说,如果多个Observables从多个线程中发射数据,必须要满足happens-before原则。

下面来看一个简单的例子:

final PublishSubject<Integer> subject = PublishSubject.create();

subject.subscribe(new Subscriber<Integer>() {
 @Override
 public void onCompleted() {

 }

 @Override
 public void onError(Throwable e) {

 }

 @Override
 public void onNext(Integer integer) {
  unSafeCount = unSafeCount + integer;
  Log.d("TAG", "onNext: " + unSafeCount);
 }
});

findViewById(R.id.send).setOnClickListener(new View.OnClickListener() {
 @Override
 public void onClick(View v) {
  final int unit = 1;
  for(int i = 0;i < 10;i++) {
   new Thread(new Runnable() {
    @Override
    public void run() {
     for (int j = 0; j < 1000; j++) {
      subject.onNext(unit);
     }
    }
   }).start();
  }
 }
});

这是一个最典型的多线程问题,从10个线程中发射数据并相加,这样最终得到的答案是小于10000的。虽然使用了RxJava,但是这样的使用对于并发是没有意义的,因为RxJava并没有去处理并发带来的问题。我们可以看下subject的onNext方法的源码,里面很简单,就是调用了对应observer的onNext方法而已。不止是这样,绝大多数的Subject都是线程不安全的,所以当你在使用这样的类的时候(典型场景就是自制的RxBus),如果从多个线程中发射数据,那你就要小心了。

对于这样的问题,有两种解决方案:

第一种就是简单的使用传统的解决方法,比如用AtomicInteger代替int。

第二种则是使用RxJava的解决方案,在这里就是用SerializedSubject去代替Subject:

final PublishSubject<Integer> subject = PublishSubject.create();

subject.subscribe(new Subscriber<Integer>() {
 @Override
 public void onCompleted() {

 }

 @Override
 public void onError(Throwable e) {

 }

 @Override
 public void onNext(Integer integer) {
  unSafeCount = unSafeCount + integer;
  count.addAndGet(integer);

  Log.d("TAG", "onNext: " + count);
 }
});

final SerializedSubject<Integer, Integer> ser = new SerializedSubject<Integer, Integer>(subject);

findViewById(R.id.send).setOnClickListener(new View.OnClickListener() {
 @Override
 public void onClick(View v) {
  final int unit = 1;

  for(int i = 0;i < 10;i++){
   new Thread(new Runnable() {
    @Override
    public void run() {
     for(int j = 0;j < 1000;j++){
      ser.onNext(unit);
     }
    }
   }).start();
  }
 }
});

可以看一下SerializedSubject的onNext方法做了什么:

@Override
public void onNext(T t) {
 if (terminated) {
  return;
 }
 synchronized (this) {
  if (terminated) {
   return;
  }
  if (emitting) {
   FastList list = queue;
   if (list == null) {
    list = new FastList();
    queue = list;
   }
   list.add(nl.next(t));
   return;
  }
  emitting = true;
 }
 try {
  actual.onNext(t);
 } catch (Throwable e) {
  terminated = true;
  Exceptions.throwOrReport(e, actual, t);
  return;
 }
 for (;;) {
  for (int i = 0; i < MAX_DRAIN_ITERATION; i++) {
   FastList list;
   synchronized (this) {
    list = queue;
    if (list == null) {
     emitting = false;
     return;
    }
    queue = null;
   }
   for (Object o : list.array) {
    if (o == null) {
     break;
    }
    try {
     if (nl.accept(actual, o)) {
      terminated = true;
      return;
     }
    } catch (Throwable e) {
     terminated = true;
     Exceptions.throwIfFatal(e);
     actual.onError(OnErrorThrowable.addValueAsLastCause(e, t));
     return;
    }
   }
  }
 }
}

处理方式很简单,如果有其他线程在发射数据,那就将数据放置到队列中,等待下次发射。这保证了同一时间只会有一个线程调用onNext,onComplete和onError这些方法。

但是这样操作显然是会造成性能的影响的,所以RxJava并不会把所有的操作都打上线程安全的标签。

在这里就要引申出一个问题,那就是使用者对create方法的滥用,其实这个方法不应该被使用者频繁的调用的,因为你必须要小心的处理所有的数据发射,接收的逻辑。相反的,使用已有的操作符能很好的解决这个问题,所以下次大家在遇到问题的时候不要简单的使用create去自己写,而是应该想想有没有现成的操作符可以完成相应的需求。

RxJava中的一些操作符

RxJava中有一些操作符也和多线程并发有关,下面让我来讲一讲merge和concat,以及他们的一些变种操作符。

对于多线程发射数据,有时候我们需要得到的结果也保持和发射时候一样的顺序,这个时候如果我们使用merge这个操作符去结合多个发射源,那么就会产生一定的问题了(例子中做了非常不好的示范——使用了create操作符,请大家不要学习这样的写法,这里单纯是为了求证结果)。

Observable o1 = Observable.create(new Observable.OnSubscribe<Integer>() {
 @Override
 public void call(final Subscriber<? super Integer> subscriber) {
  new Thread(new Runnable() {
   @Override
   public void run() {
    try {
     Thread.sleep(1000);
     subscriber.onNext(1);
     subscriber.onCompleted();
    } catch (InterruptedException e) {
     e.printStackTrace();
    }
   }
  }).start();
 }
});
Observable o2 = Observable.create(new Observable.OnSubscribe<Integer>() {
 @Override
 public void call(Subscriber<? super Integer> subscriber) {
  subscriber.onNext(2);
  subscriber.onCompleted();
 }
});

Observable.merge(o1,o2)
  .subscribe(new Subscriber<Integer>() {
   @Override
   public void onCompleted() {

   }

   @Override
   public void onError(Throwable e) {

   }

   @Override
   public void onNext(Integer i) {
    Log.d("TAG", "onNext: " + i);
   }
  });

对于这样的场景,我们得到的答案将是2,1而不是先得到o1发射的数据,再获取o2的数据。

究其原因,就是因为merge其实就是给什么传什么,也不会去管数据发射的顺序:

@Override
public void onNext(Observable<? extends T> t) {
  if (t == null) {
    return;
  }
  if (t == Observable.empty()) {
    emitEmpty();
  } else
  if (t instanceof ScalarSynchronousObservable) {
    tryEmit(((ScalarSynchronousObservable<? extends T>)t).get());
  } else {
    InnerSubscriber<T> inner = new InnerSubscriber<T>(this, uniqueId++);
    addInner(inner);
    t.unsafeSubscribe(inner);
    emit();
  }
}

可以看到在经过lift操作之后,对应的中间人MergeSubscriber的onNext,没有什么多余的代码,所以在多个Observable从多线程中发射数据的时候,顺序当然不能得到保证。

一个单词说明这个问题:interleaving——交错。merge后的数据源可能是交错的。由于merge有这样数据交错的问题,所以它的变种—flatMap也会有同样的问题。

对于这样的场景,我们可以使用concat操作符来完成:

Concat waits to subscribe to each additional Observable that you pass to it until the previous Observable completes.

根据文档,我们知道concat操作符是一个接一个的处理数据源的数据的。

if (wip.getAndIncrement() != 0) {
  return;
}

final int delayErrorMode = this.delayErrorMode;

for (;;) {
  if (actual.isUnsubscribed()) {
    return;
  }

  if (!active) {
    if (delayErrorMode == BOUNDARY) {
      if (error.get() != null) {
        Throwable ex = ExceptionsUtils.terminate(error);
        if (!ExceptionsUtils.isTerminated(ex)) {
          actual.onError(ex);
        }
        return;
      }
    }

    boolean mainDone = done;
    Object v = queue.poll();
    boolean empty = v == null;

    if (mainDone && empty) {
      Throwable ex = ExceptionsUtils.terminate(error);
      if (ex == null) {
        actual.onCompleted();
      } else
      if (!ExceptionsUtils.isTerminated(ex)) {
        actual.onError(ex);
      }
      return;
    }

    if (!empty) {

      Observable<? extends R> source;

      try {
        source = mapper.call(NotificationLite.<T>instance().getValue(v));
      } catch (Throwable mapperError) {
        Exceptions.throwIfFatal(mapperError);
        drainError(mapperError);
        return;
      }

      if (source == null) {
        drainError(new NullPointerException("The source returned by the mapper was null"));
        return;
      }

      if (source != Observable.empty()) {

        if (source instanceof ScalarSynchronousObservable) {
          ScalarSynchronousObservable<? extends R> scalarSource = (ScalarSynchronousObservable<? extends R>) source;

          active = true;

          arbiter.setProducer(new ConcatMapInnerScalarProducer<T, R>(scalarSource.get(), this));
        } else {
          ConcatMapInnerSubscriber<T, R> innerSubscriber = new ConcatMapInnerSubscriber<T, R>(this);
          inner.set(innerSubscriber);

          if (!innerSubscriber.isUnsubscribed()) {
            active = true;

            source.unsafeSubscribe(innerSubscriber);
          } else {
            return;
          }
        }
        request(1);
      } else {
        request(1);
        continue;
      }
    }
  }
  if (wip.decrementAndGet() == 0) {
    break;
  }
}

通过源码我们可以知道,active字段就保证了如果上一个数据源还没有发射完数据,就会一直在for循环中等待,直到上一个数据源发射完了数据重置了active字段。

对于concat,其实还存在一个问题,那就是多个Observable变成了串行,会大大的增加整个RxJava事件流的处理时间,对于这个场景,我们可以使用concatEager来解决。concatEager的源码就不带大家分析了,有兴趣的同学可以自行查看。

总结

这篇文章比较短,讲的东西也比较浅显,其实就是讨论了一下RxJava中多线程并发的几个问题。最后我想说,RxJava并不是什么高大上的东西,在你的项目引入之前,要考虑一下是否真的有必要这么做。就算真的有场景需要RxJava,也请不要一口气把项目中所有的操作都换成RxJava,一些简单的操作不一定需要使用RxJava的操作符的实现,用了反而降低了代码的可读性,切勿为了使用Rx而使用Rx。

好了,以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作能带来一定的帮助,如果有疑问大家可以留言交流。

时间: 2017-02-28

Java 多线程有序执行的几种方法总结

Java 多线程有序执行的几种方法总结 同事无意间提出了这个问题,亲自实践了两种方法.当然肯定还会有更多更好的方法. 方法一 import java.util.concurrent.atomic.AtomicInteger; public class OrderedThread1 { static AtomicInteger count = new AtomicInteger(0); public static void main(String[] args) throws Interrupte

Java编程之多线程死锁与线程间通信简单实现代码

死锁定义 死锁是指两个或者多个线程被永久阻塞的一种局面,产生的前提是要有两个或两个以上的线程,并且来操作两个或者多个以上的共同资源:我的理解是用两个线程来举例,现有线程A和B同时操作两个共同资源a和b,A操作a的时候上锁LockA,继续执行的时候,A还需要LockB进行下面的操作,这个时候b资源在被B线程操作,刚好被上了锁LockB,假如此时线程B刚好释放了LockB则没有问题,但没有释放LockB锁的时候,线程A和B形成了对LockB锁资源的争夺,从而造成阻塞,形成死锁:具体其死锁代码如下:

Java实现多线程文件下载的代码示例

Java实现多线程文件下载思路: 1.基本思路是将文件分段切割.分段传输.分段保存. 2.分段切割用到HttpUrlConnection对象的setRequestProperty("Range", "bytes=" + start + "-" + end)方法. 3.分段传输用到HttpUrlConnection对象的getInputStream()方法. 4.分段保存用到RandomAccessFile的seek(int start)方法. 5

java实现多线程之定时器任务

在Java中Timer是java.util包中的一个工具类,提供了定时器的功能.我们可以创建一个Timer对象,然后调用其schedule方法在某个特定的时间去执行一个特定的任务.并且你可以让其以特定频率一直执行某个任务,这个任务是用TimerTask来描述的,我们只需要将要进行的操作写在TimerTask类的run方法中即可.先附上两个小例子一遍让读者了解什么是定时器.接着再分析其中的一些源码实现. 第一个小例子: package com.zkn.newlearn.thread; import

Java多线程下载文件实例详解

本文实例为大家分享了Java多线程下载文件的具体代码,供大家参考,具体内容如下 import java.io.File; import java.io.InputStream; import java.io.RandomAccessFile; import java.net.HttpURLConnection; import java.net.URL; public class MulThreadDownload { public static void main(String[] args)

java 读取本地文件实例详解

java 读取本地文件实例详解 用javax.xml.w3c解析 实例代码: package cn.com.xinli.monitor.utils; import org.w3c.dom.Document; import org.w3c.dom.Element; import javax.xml.parsers.DocumentBuilder; import javax.xml.parsers.DocumentBuilderFactory; import java.io.File; /** *

Java多线程用法的实例详解

Java多线程用法的实例详解 前言: 最全面的java多线程用法解析,如果你对Java的多线程机制并没有深入的研究,那么本文可以帮助你更透彻地理解Java多线程的原理以及使用方法. 1.创建线程 在Java中创建线程有两种方法:使用Thread类和使用Runnable接口.在使用Runnable接口时需要建立一个Thread实例.因此,无论是通过Thread类还是Runnable接口建立线程,都必须建立Thread类或它的子类的实例.Thread构造函数: public Thread( ); p

Java生成CSV文件实例详解

本文实例主要讲述了Java生成CSV文件的方法,具体实现步骤如下: 1.新建CSVUtils.java文件: package com.saicfc.pmpf.internal.manage.utils; import java.io.BufferedWriter; import java.io.File; import java.io.FileInputStream; import java.io.FileNotFoundException; import java.io.FileOutputS

Java多线程 中断机制及实例详解

正文 这里详细分析interrupt(),interrupted(),isInterrupted()三个方法 interrupt() 中断这个线程,设置中断标识位 public void interrupt() { if (this != Thread.currentThread()) checkAccess(); synchronized (blockerLock) { Interruptible b = blocker; if (b != null) { interrupt0(); // J

FasfDFS整合Java实现文件上传下载功能实例详解

在上篇文章给大家介绍了FastDFS安装和配置整合Nginx-1.13.3的方法,大家可以点击查看下. 今天使用Java代码实现文件的上传和下载.对此作者提供了Java API支持,下载fastdfs-client-java将源码添加到项目中.或者在Maven项目pom.xml文件中添加依赖 <dependency> <groupId>org.csource</groupId> <artifactId>fastdfs-client-java</arti

jsp按格式导出doc文件实例详解

jsp按格式导出doc文件实例详解 原理:doc文件其实可以保存为xml文件,该xml文件用字符串表示了doc文件的表现形式,我们只需要用Java将那些要填的内容替换掉然后下载给客户就行了. 1.首先是按照你的文档填写好数据. 2.将文档另存为xml文件,然后编辑该xml文件,将填好的内容用某种格式替换,如:将名字张三替换成${name} 3.读取文件,将文件中的${name}替换成真正的名字. 4.下载. 接下来看代码: 首先是那个转换类 package com.my.util; import

Java 实现FTP服务实例详解

Java 实现FTP服务实例详解 1.FTP简介 FTP 是File Transfer Protocol(文件传输协议)的英文简称,而中文简称为"文传协议".用于Internet上的控制文件的双向传输.同时,它也是一个应用程序(Application).基于不同的操作系统有不同的FTP应用程序,而所有这些应用程序都遵守同一种协议以传输文件.在FTP的使用当中,用户经常遇到两个概念:"下载"(Download)和"上传"(Upload)."

java 代理机制的实例详解

java 代理机制的实例详解 前言: java代理分静态代理和动态代理,动态代理有jdk代理和cglib代理两种,在运行时生成新的子类class文件.本文主要练习下动态代理,代码用于备忘.对于代理的原理和机制,网上有很多写的很好的,就不班门弄斧了. jdk代理 实例代码 import java.lang.reflect.InvocationHandler; import java.lang.reflect.Method; import java.lang.reflect.Proxy; publi

Mybatis映射文件实例详解

 一.输入映射 parameterType 指定输入参数的Java类型,可以使用别名或者类的全限定名.它可以接收简单类型.POJO.HashMap. 1.传递简单类型 根据用户ID查询用户信息: <select id="findUserById" parameterType="int" resultType="com.itheima.mybatis.po.User"> SELECT * FROM USER WHERE id =#{id