ReactiveX是专注于异步工作的API，它将异步事件的处理与观察者模式、迭代器模式及函数式编程相结合了起来。实时地处理返回数据是在工程中经常出现的情景，所以使用高效、可拓展的方式来解决这种问题非常重要。ReactiveX通过观察者模式以及操作符来提供灵活地处理异步通信的方式，你不用再去关注线程创造与同步这些繁琐的事情。
RxJava是一个开源的实现ReactiveX的工具。这里面有两种主要的类：Observalbe和Subscriber。在RxJava中，Observable类产生异步数据或事件，Subscriber类对这些数据和事件进行操作。正常的工作流程就是Observable产生一系列的数据或事件，然后完成或者产生异常。一个Observable可以拥有多个Subscriber，每一个被生成的事件都会触发被绑定的Subscriber的onNext()方法。当一个Observable生成完所有事件后，所有绑定的Subscriber的onCompleted()方法会被调用（在发生异常时会调用onError()方法）。现在我们对这两个类有了初步的认识，可以开始了解如何创建与订阅一个Observer了：
Observable integerObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext(1);
subscriber.onNext(2);
subscriber.onNext(3);
subscriber.onCompleted();
}
});我们创建的这个Observable产生了1、2、3三条数据，现在需要创建Subscriber来处理它们：
Subscriber integerSubscriber = new Subscriber() {
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer value) {
System.out.println("onNext: " + value);
}
};你可以看到，这个Subscriber将每一个收到的数据打印了出来。当你创建完Observable与Subscriber后，你需要将它们用Observable.subscribe()方法连接起来。
integerObservable.subscribe(integerSubscriber);
// 输出:
// onNext: 1
// onNext: 2
// onNext: 3
// Complete!这整个过程，可以使用Observable.just()来简化数据的生成过程，再创建Subscriber的匿名内部类来简化处理过程：
Observable.just(1, 2 ,3).subscribe(new Subscriber() {
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(Integer value) {
System.out.println("onNext: " + value);
}
});只是创建与订阅Observable非常简单，然而这只是RxJava的冰山一角。任何Observable都可以将它的输出通过“操作符”来进行预处理，一个Observable可以绑定多个操作符来进行链式处理。举个例子，前面使用的Observable只是输出简单的数字，但是我们只想对奇数进行处理，可以通过设置filter来实现：
Observable.just(1, 2, 3, 4, 5, 6) // add more numbers
.filter(new Func1() {
@Override
public Boolean call(Integer value) {
return value % 2 == 1;
}
})
.subscribe(new Subscriber() {
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) {
}
@Override
public void onNext(Integer value) {
System.out.println("onNext: " + value);
}
});
// Outputs:
// onNext: 1
// onNext: 3
// onNext: 5
// Complete!filter()操作符定义了一个函数，它只会在输出为奇数的时候返回true，输出为偶数的时候返回false。返回false的输出是不会发送到Subscriber的。值得注意的是filter函数返回了一个Observable，我们可以进行链式处理。比如现在希望找到所有奇数的平方根，你可以在Subscriber获取到数字的时候来处理这个逻辑，但是这样的话就不能对做平方根后的数字进行进一步处理了。你可以在filter操作符后加一个map操作符来实现这个逻辑：
Observable.just(1, 2, 3, 4, 5, 6) // add more numbers
.filter(new Func1() {
@Override
public Boolean call(Integer value) {
return value % 2 == 1;
}
})
.map(new Func1() {
@Override
public Double call(Integer value) {
return Math.sqrt(value);
}
})
.subscribe(new Subscriber() { // notice Subscriber type changed to
@Override
public void onCompleted() {
System.out.println("Complete!");
}
@Override
public void onError(Throwable e) { }
@Override
public void onNext(Double value) {
System.out.println("onNext: " + value);
}
});
// Outputs:
// onNext: 1.0
// onNext: 1.7320508075688772
// onNext: 2.23606797749979
// Complete!链式操作符是RxJava不可分割的一块内容，它简便、可拓展，功能强大。现在你对Observable与Subscriber的联系有了进一步的认识，我们可以开始下一个话题：将RxJava与Android开发结合起来。
在Android开发中，我们常常需要在后台线程中完成一些耗时工作，当后台线程完成工作之后通知主线程来显示结果。Android Sdk中有很多类可以完成多这种工作，Asynctask、Loader、Service等等。然而它们有时候并不是最好的选择。Asynctask经常会造成内存泄露；与ContentProvider搭配的CursorLoader使用起来非常繁琐，并且需要大量样板代码；Service通常是用于执行需要长时间在后台运行的任务。那么我们来看看RxJava怎么来解决这些问题。
在下面这个布局中，有一个按钮，它会启动一个耗时任务，并且将任务的进度展示到ProgressBar上。
<LinearLayout xmlns:android="http://schemas.android.com/apk/res/android"
xmlns:app="http://schemas.android.com/apk/res-auto"
android:id="@+id/root_view"
android:layout_width="match_parent"
android:layout_height="match_parent"
android:fitsSystemWindows="true"
android:orientation="vertical">
<android.support.v7.widget.Toolbar
android:id="@+id/toolbar"
android:layout_width="match_parent"
android:layout_height="?attr/actionBarSize"
android:background="?attr/colorPrimary"
app:popupTheme="@style/AppTheme.PopupOverlay"
app:theme="@style/ThemeOverlay.AppCompat.Dark.ActionBar" />
<Button
android:id="@+id/start_btn"
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_gravity="center_horizontal"
android:text="@string/start_operation_text" />
<ProgressBar
android:layout_width="wrap_content"
android:layout_height="wrap_content"
android:layout_gravity="center_horizontal"
android:indeterminate="true" />
</LinearLayout>一旦按钮按下之后，我们禁用按钮，同时在后台线程中开始做耗时任务，当任务结束之后会弹出一个SnackBar。现在使用一个简单的AsyncTask来完成如下这个耗时任务longRunningOperation()。
public String longRunningOperation() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
// error
}
return "Complete!";
}
private class SampleAsyncTask extends AsyncTask {
@Override
protected String doInBackground(Void... params) {
return longRunningOperation();
}
@Override
protected void onPostExecute(String result) {
Snackbar.make(rootView, result, Snackbar.LENGTH_LONG).show();
startAsyncTaskButton.setEnabled(true);
}
}现在要怎么用RxJava替换掉呢？首先需要在app模块的gradle脚本中添加依赖：io.reactivex:rxjava:1.0.14。然后创建一个Observable来运行耗时任务，调用Observable.create方法来实现这个功能：
final Observable operationObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext(longRunningOperation());
subscriber.onCompleted();
}
});现在创建的这个Observable会执行longRunningOperation()，在执行完毕后通知Subscriber，并且调用Subscriber的onCompleted()方法（注意：耗时任务在Subscriber与Observable绑定之前不会被执行）。接下来，我们需要在按钮点击事件上将Observable与Subscriber绑定。
startRxOperationButton.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(final View v) {
v.setEnabled(false);
operationObservable.subscribe(new Subscriber() {
@Override
public void onCompleted() {
v.setEnabled(true);
}
@Override
public void onError(Throwable e) {}
@Override
public void onNext(String value) {
Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show();
}
});
}
});接下来我们就可以运行这个应用了。但是你会发现，点击按钮之后ProgressBar并不会有反应！（这是因为UI主线程被耗时任务阻塞了）。我们还没有指定Observable的运行线程，也没有指定Subscriber的运行线程。RxJava调度的功能就在这里体现：我们可以指定两个不同的线程，一个是运行线程，一个是监视线程。使用Observable.observeOn()函数，你可以定义一个线程用来监视Observable运行，并且查看是否Observable有新的输出（Subscriber的onNext，onCompleted， onError方法都是在这个监视线程中执行的）。调用Observable.subscribeOn()方法可以指定一个线程运行耗时任务。RxJava原本将两个任务都安排在一个线程中执行，而你可以通过调用observeOn与subcribeOn方法来进行多线程操作。RxJava封装了很多调度工具，比如Schedulers.io()（阻塞I/O操作），Schedulers.computation()（进行计算工作），Schedulers.newThread()（创建新线程）。不过在Android开发中，你应该关心的是怎么将代码执行在主线程中。我们可以通过RxAndroid库来完成这个工作！
RxAndroid是RxJava的轻量级拓展工具，它提供了运行在主线程上的Scheduler，或者运行在任意Handler线程上的Scheduler！这样一来，我们就可以在后台线程中运行耗时工作，同时在主线程中处理结果。要使用RxAndroid，你需要添加依赖io.reactivex:rxandroid:1.0.1。
final Observable operationObservable = Observable.create(new Observable.OnSubscribe() {
@Override
public void call(Subscriber subscriber) {
subscriber.onNext(longRunningOperation());
subscriber.onCompleted();
}
}).subscribeOn(Schedulers.io()) // subscribeOn the I/O thread
.observeOn(AndroidSchedulers.mainThread()); // observeOn the UI Thread以上经过修改代码会在Schedulers.io线程上运行耗时工作，Subscriber会在主线程中处理输出结果。现在我们运行应用，点击按钮，我们的耗时任务没有阻塞UI线程。RxJava 1.0.13同时提供了Single类，它只会产生一个结果，之后马上执行onComplete()方法。示例如下：
Subscription subscription = Single.create(new Single.OnSubscribe() {
@Override
public void call(SingleSubscriber singleSubscriber) {
String value = longRunningOperation();
singleSubscriber.onSuccess(value);
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1() {
@Override
public void call(String value) {
// onSuccess
Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show();
}
}, new Action1() {
@Override
public void call(Throwable throwable) {
// handle onError
}
});在订阅Single的时候，只有onSuccess()、onError()可以监听。同时你可以调用Single.mergeWith()操作符来将多个Single合成一个Observable，这个被合成的Observable会依次输出所有Single的结果。
我们之前提到，使用AsyncTask的一大缺点就是它可能会造成内存泄露。当它们持有的Activity/Fragment的引用没有正确处理时就会这样。不幸的是，RxJava并不会自动防止这种情况发生，好在它可以很容易地防止内存泄露。Observable.subscribe()方法会返回一个Subscription对象，这个对象仅仅有两个方法：isSbscribed()与unsubscribe()。你可以在Activity/Fragment的onDestroy方法中调用Subscription.isSubscribed()检测是否这个异步任务仍在进行。如果它仍在进行，则调用unsubscribe()方法来结束任务，从而释放其中的强引用，防止内存泄露。如果你使用了多个Observable与Subscriber，那么你可以将它们添加到CompositeSubscription中，并调用CompositeSubscription.unsubscribe()结束所有的任务。
RxJava为Android提供了一个非常棒的多线程同步的解决方案。在Android开发中，让耗时任务在后台线程运行，并将结果在主线程中展示，是非常重要的。同时它的操作符模式为这个工具起到了锦上添花的作用。使用RxJava的时候，你需要对整个工具库有一定理解，才能将其发挥最大的作用。所以花时间去系统地学习这个库是非常值得的。未来我还将会在我的博客中讨论RxJava的几个特性：冷热Observable的比较、处理反压力、Rx子类。本节中的样本代码可以从我的Github中找到。
Java 8 引入了Lambda表达式，可惜的是Android不支持Java 8，所以我们不能使用RxJava与这一特性结合的功能。幸运的是，有一个工具叫RetroLambda可以将Lambda表达式与低版本的Java兼容。同时还有一个Gradle插件可以在Android开发中使用。使用了Lambda表达式，你的RxJava代码会更加简洁：
final Observable operationObservable = Observable.create(
(Subscriber subscriber) -> {
subscriber.onNext(longRunningOperation());
subscriber.onCompleted();
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread());
startRxOperationButton = (Button) findViewById(R.id.start_rxjava_operation_btn);
startRxOperationButton.setOnClickListener(v -> {
v.setEnabled(false);
operationObservable.subscribe(
value -> Snackbar.make(rootView, value, Snackbar.LENGTH_LONG).show(),
error -> Log.e("TAG", "Error: " + error.getMessage()),
() -> v.setEnabled(true));
});Lambda表达式减少了很多RxJava的样本代码，我强烈推荐将RetroLambda引入工程中使用。它给开发带来的好处不仅仅于RxJava上（你会注意到OnCliclListener也是用Lambda表达式设置的）!
京东创始人刘强东和其妻子章泽天最近成为了互联网舆论关注的焦点。有关他们“移民美国”和在美国购买豪宅的传言在互联网上广泛传播。然而，京东官方通过微博发言人发布的消息澄清了这些传言，称这些言论纯属虚假信息和蓄意捏造。
日前，据博主“@超能数码君老周”爆料，国内三大运营商中国移动、中国电信和中国联通预计将集体采购百万台规模的华为Mate60系列手机。
据报道，荷兰半导体设备公司ASML正看到美国对华遏制政策的负面影响。阿斯麦（ASML）CEO彼得·温宁克在一档电视节目中分享了他对中国大陆问题以及该公司面临的出口管制和保护主义的看法。彼得曾在多个场合表达了他对出口管制以及中荷经济关系的担忧。
今年早些时候，抖音悄然上线了一款名为“青桃”的 App，Slogan 为“看见你的热爱”，根据应用介绍可知，“青桃”是一个属于年轻人的兴趣知识视频平台，由抖音官方出品的中长视频关联版本，整体风格有些类似B站。
日前，威马汽车首席数据官梅松林转发了一份“世界各国地区拥车率排行榜”，同时，他发文表示：中国汽车普及率低于非洲国家尼日利亚，每百户家庭仅17户有车。意大利世界排名第一，每十户中九户有车。
近日，一项新的研究发现，维生素 C 和 E 等抗氧化剂会激活一种机制，刺激癌症肿瘤中新血管的生长，帮助它们生长和扩散。
据媒体援引消息人士报道，苹果公司正在测试使用3D打印技术来生产其智能手表的钢质底盘。消息传出后，3D系统一度大涨超10%，不过截至周三收盘，该股涨幅回落至2%以内。
9月2日，坐拥千万粉丝的网红主播“秀才”账号被封禁，在社交媒体平台上引发热议。平台相关负责人表示，“秀才”账号违反平台相关规定，已封禁。据知情人士透露，秀才近期被举报存在违法行为，这可能是他被封禁的部分原因。据悉，“秀才”年龄39岁，是安徽省亳州市蒙城县人，抖音网红，粉丝数量超1200万。他曾被称为“中老年...
9月3日消息，亚马逊的一些股东，包括持有该公司股票的一家养老基金，日前对亚马逊、其创始人贝索斯和其董事会提起诉讼，指控他们在为 Project Kuiper 卫星星座项目购买发射服务时“违反了信义义务”。
据消息，为推广自家应用，苹果现推出了一个名为“Apps by Apple”的网站，展示了苹果为旗下产品（如 iPhone、iPad、Apple Watch、Mac 和 Apple TV）开发的各种应用程序。
特斯拉本周在美国大幅下调Model S和X售价，引发了该公司一些最坚定支持者的不满。知名特斯拉多头、未来基金（Future Fund）管理合伙人加里·布莱克发帖称，降价是一种“短期麻醉剂”，会让潜在客户等待进一步降价。
据外媒9月2日报道，荷兰半导体设备制造商阿斯麦称，尽管荷兰政府颁布的半导体设备出口管制新规9月正式生效，但该公司已获得在2023年底以前向中国运送受限制芯片制造机器的许可。
近日，根据美国证券交易委员会的文件显示，苹果卫星服务提供商 Globalstar 近期向马斯克旗下的 SpaceX 支付 6400 万美元（约 4.65 亿元人民币）。用于在 2023-2025 年期间，发射卫星，进一步扩展苹果 iPhone 系列的 SOS 卫星服务。
据报道，马斯克旗下社交平台𝕏（推特）日前调整了隐私政策，允许 𝕏 使用用户发布的信息来训练其人工智能（AI）模型。新的隐私政策将于 9 月 29 日生效。新政策规定，𝕏可能会使用所收集到的平台信息和公开可用的信息，来帮助训练 𝕏 的机器学习或人工智能模型。
9月2日，荣耀CEO赵明在采访中谈及华为手机回归时表示，替老同事们高兴，觉得手机行业，由于华为的回归，让竞争充满了更多的可能性和更多的魅力，对行业来说也是件好事。
《自然》30日发表的一篇论文报道了一个名为Swift的人工智能（AI）系统，该系统驾驶无人机的能力可在真实世界中一对一冠军赛里战胜人类对手。
近日，非营利组织纽约真菌学会（NYMS）发出警告，表示亚马逊为代表的电商平台上，充斥着各种AI生成的蘑菇觅食科普书籍，其中存在诸多错误。
社交媒体平台𝕏（原推特）新隐私政策提到：“在您同意的情况下，我们可能出于安全、安保和身份识别目的收集和使用您的生物识别信息。”
2023年德国柏林消费电子展上，各大企业都带来了最新的理念和产品，而高端化、本土化的中国产品正在不断吸引欧洲等国际市场的目光。
罗永浩日前在直播中吐槽苹果即将推出的 iPhone 新品，具体内容为：“以我对我‘子公司’的了解，我认为 iPhone 15 跟 iPhone 14 不会有什么区别的，除了序（列）号变了，这个‘不要脸’的东西，这个‘臭厨子’。
