java9發(fā)布以來(lái)一直備受關(guān)注的是帶來(lái)最大變化的模塊化。模塊化可以說(shuō)是java9的核心思想挠乳。前面也大概介紹了java9的模塊化。這次來(lái)介紹下java9的另一個(gè)新特性讼呢，異步非阻塞的反向控制流撩鹿。
反向控制流
下面就是Java9的反向控制流的大概處理過(guò)程。
Flow介紹
java9中提供了java.util.concurrent.Flow類悦屏，里面有很多內(nèi)部接口類节沦。比如：
- Publisher 發(fā)布者键思。
- Subscriber 訂閱者。
- Subscription 處理發(fā)布者/訂閱者的關(guān)系類甫贯。
- Processor 包含了發(fā)布者&訂閱者吼鳞。
|方法名|參數(shù)|描述|
|--------|:--:|:------------:|----|
|Publisher.subscribe|Subscriber|給當(dāng)前發(fā)布者綁定一個(gè)訂閱者。|
|Subscriber.onSubscribe|Subscription|當(dāng)訂閱者訂閱了一個(gè)發(fā)布者的時(shí)候叫搁，需要執(zhí)行的后續(xù)操作赔桌。
|Subscriber.onNext|Object|訂閱者主動(dòng)向發(fā)布者請(qǐng)求數(shù)據(jù)后，發(fā)布者把數(shù)據(jù)發(fā)布出來(lái)渴逻，然后Subscription按次將數(shù)據(jù)丟到onNext中疾党。訂閱者這時(shí)可以按照自己的業(yè)務(wù)處理數(shù)據(jù)。
|Subscriber.onError|Throwable|當(dāng)訂閱者請(qǐng)求的數(shù)據(jù)長(zhǎng)度<=0時(shí)惨奕，或者其它異常的時(shí)候雪位，Subscription會(huì)調(diào)用onError方法，訂閱者需要進(jìn)行錯(cuò)誤處理梨撞。
|Subscriber.onComplete|-|在數(shù)據(jù)處理完成后執(zhí)行onComplete方法雹洗，訂閱者可以進(jìn)行完成的后續(xù)動(dòng)作。
|Subscription.request|long|通過(guò)主動(dòng)發(fā)起請(qǐng)求數(shù)據(jù)的方式卧波，獲取N個(gè)數(shù)據(jù)在當(dāng)前的訂閱中时肿，并且順次的調(diào)用訂閱者的onNext方法。其中N是控制數(shù)據(jù)流的流速幽勒。如果傳入Long.MAX_VALUE則表示無(wú)限制的流速嗜侮。發(fā)布者將會(huì)一次把所有的數(shù)據(jù)都取到當(dāng)次訂閱Subscription中，由Subscription通知到訂閱者啥容。
|Subscription.cancle|-|取消接受消息锈颗，該方法不會(huì)觸發(fā)onComplete或者onError|
SubmissionPublisher
java9已經(jīng)實(shí)現(xiàn)了一套發(fā)布，即SubmissionPublisher咪惠，我們只需要實(shí)現(xiàn)自己的訂閱業(yè)務(wù)邏輯即可击吱。demo代碼如下：
public class MySubscribe<T> implements Flow.Subscriber<T> {
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
this.subscription = subscription;
}
@Override
public void onNext(T item) {
System.out.println("onNext get: " + item);
subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("Done.");
}
public void start(int n){
subscription.request(n);
}
}
public static void main(String[] args) throws InterruptedException {
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
MySubscribe<String> subscribe = new MySubscribe<>();
publisher.subscribe(subscribe);//產(chǎn)生發(fā)布訂閱關(guān)系
List.of("1","2","3").forEach(publisher::submit);//將數(shù)據(jù)添加到發(fā)布者中。
subscribe.start(1);//手動(dòng)開(kāi)始訂閱遥昧。
//需要關(guān)閉publisher覆醇，否則無(wú)論如何都不會(huì)執(zhí)行onComplete方法
//關(guān)閉不分先后，關(guān)閉后仍然可以獲取數(shù)據(jù)炭臭，但是不能往發(fā)布者中submit數(shù)據(jù)了永脓。
publisher.close();
//由于是異步非阻塞的，這邊把線程睡眠2s等待異步執(zhí)行完成鞋仍。
Thread.sleep(2000);
}
最后執(zhí)行的結(jié)果如下：
onNext get: 1
onNext get: 2
onNext get: 3
Done.
上面示例是手動(dòng)開(kāi)始請(qǐng)求數(shù)據(jù)常摧，也可以在產(chǎn)生訂閱關(guān)系的時(shí)候就開(kāi)始請(qǐng)求獲取數(shù)據(jù)。在onSubscribe中添加subscription.request(1);。
Processor
如果遇到多個(gè)數(shù)據(jù)處理流的話落午，就可以用Processor谎懦。它即使一個(gè)發(fā)布者也是一個(gè)訂閱者（它訂閱上一個(gè)發(fā)布者，并且可以作為下一個(gè)訂閱者的發(fā)布者）溃斋。下面分別是一個(gè)類型轉(zhuǎn)換處理和偶數(shù)過(guò)濾器：
public class EvenFilterProcessor extends SubmissionPublisher<Integer> implements Flow.Processor<Integer, Integer>{
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
this.subscription = subscription;
subscription.request(1);
}
@Override
public void onNext(Integer item) {
if(item % 2 == 0) {
submit(item);//將數(shù)據(jù)提交到下一個(gè)發(fā)布者中界拦。
}
subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
//do something
}
@Override
public void onComplete() {
close();//注意多個(gè)發(fā)布訂閱鏈需要在完成的時(shí)候調(diào)用。通知下一個(gè)發(fā)布者完成梗劫。
}
}
public class StringToIntegerProcessor extends SubmissionPublisher<Integer> implements Flow.Processor<String, Integer>{
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
this.subscription = subscription;
subscription.request(1);
}
@Override
public void onNext(String item) {
submit(Integer.valueOf(item));
subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
//do something
}
@Override
public void onComplete() {
close();
}
}
具體調(diào)用方法：
public static void main(String[] args) throws InterruptedException {
SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
StringToIntegerProcessor convert = new StringToIntegerProcessor();
publisher.subscribe(convert);
EvenFilterProcessor filter = new EvenFilterProcessor();
convert.subscribe(filter);
MySubscribe<Integer> subscribe = new MySubscribe<>();
filter.subscribe(subscribe);
List.of("1","2","3","4","5","6","8").forEach(publisher::submit);
publisher.close();
subscribe.start(1);
Thread.sleep(2000);
}
輸出結(jié)果：
onNext get: 2
onNext get: 4
onNext get: 6
onNext get: 8
Done.
小結(jié)
java9的這個(gè)新特性使代碼寫(xiě)起來(lái)更簡(jiǎn)潔享甸，更易懂，而且維護(hù)起來(lái)更方便在跳。至于性能方面枪萄，這里面沒(méi)有去測(cè)試，但是是異步非阻塞的線程猫妙，應(yīng)該性能不錯(cuò)瓷翻。
