JDK8中提供了流式对数据进行处理的功能,它的出现允许我们以声明式方式对数据集合进行处理。所谓声明式是相对于我们平时所用的命令式编程来说的,使用声明式编程会让我们对业务的表达更清晰。另外使用流可以让我们很方便地对数据集进行并行处理。
比如下面的代码,我们从person列表中过滤出年龄大于10岁的人,并且收集对应的name字段到list,然后统一打印处理。在使用非Stream的情况下,我们会使用如下代码来实现。
public static List<Person> makeList() {
List<Person> personList = new ArrayList<Person>();
Person p1 = new Person();
p1.setAge(10);
p1.setName("zlx");
personList.add(p1);
p1 = new Person();
p1.setAge(12);
p1.setName("jiaduo");
personList.add(p1);
p1 = new Person();
p1.setAge(5);
p1.setName("ruoran");
personList.add(p1);
return personList;
}
public static void noStream(List<Person> personList) {
List<String> nameList = new ArrayList<>();
for (Person person : personList) {
if (person.age >= 10) {
nameList.add(person.getName());
}
}
for(String name: nameList) {
System.out.println(name);
}
}
public static void main(String[] args) {
List<Person> personList = makeList();
noStream(personList);
}
从上述代码可知,noStream方法是典型的命令式编码,我们用for循环来一个个判断当前person对象中的age字段值是否大于等于10,如果是则把当前对象的name字段放到手动创建的nameList列表中,然后再开启新的for循环逐个遍历nameList中的name字段。
下面我们使用Stream方式来修改上面的代码。
public static void useStream(List<Person> personList) {
List<String> nameList = personList.stream().filter(person -> person.getAge() >= 10)// 1.过滤大于等于10的age字段值
.map(person -> person.getName())// 2.使用map映射元素
.collect(Collectors.toList());// 3.收集映射后元素
nameList.stream().forEach(name -> System.out.println(name));
}
在上面的代码中我们首先从personList获取到流对象,然后在其上进行了filter运算,过滤出年龄大于等于10的person,然后运用map方法映射person对象到name字段,再使用collect方法收集所有的name字段到nameList,最后从nameList上获取流并调用forEach进行打印。
上面的代码就是声明式编程,其可读性很强,代码直接可以说明想要什么(从代码就可以知道我们要过滤出年龄大于等于10岁的人,并且把满足条件的person的name字段收集起来,然后打印)。
需要注意的是,这里的filter和map操作是中间操作符,也就是当我们在流上施加这些操作时并不会真的被执行。而collect操作是终端操作符,当在流上执行终端操作符时,流上施加的操作才会执行。
JDK8中对于Steam提供了很多操作符,只是简单的列出了filter、map、collect这几种方法。
下面我们来看看当Stream与CompletableFuture相结合时会产生什么样的火花。
首先我们来看一个需求,这个需求是消费端对服务提供方集群中的某个服务进行广播调用(轮询调用同一个服务的不同提供者的机器),正常同步调用代码如下所示。
public class StreamTestFuture {
public static String rpcCall(String ip, String param) {
System.out.println(ip + " rpcCall:" + param);
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
return param;
}
public static void main(String[] args) {
// 1.生成ip列表
List<String> ipList = new ArrayList<String>();
for (int i = 1; i <=10; ++i) {
ipList.add("192.168.0." + i);
}
// 2.发起广播调用
long start = System.currentTimeMillis();
List<String> result = new ArrayList<>();
for (String ip : ipList) {
result.add(rpcCall(ip, ip));
}
// 3.输出
result.stream().forEach(r -> System.out.println(r));
System.out.println("cost:" + (System.currentTimeMillis() - start));
}
下面我们借用Stream和CompletableFuture来看看业务线程如何并发地发起多次rpc请求,从而缩短整个处理流程的耗时。
// 1.生成ip列表
List<String> ipList = new ArrayList<String>();
for (int i = 1; i <= 10; ++i) {
ipList.add("192.168.0." + i);
}
// 2.并发调用
long start = System.currentTimeMillis();
List<CompletableFuture<String>> futureList = ipList.stream()
.map(ip -> CompletableFuture.supplyAsync(() -> rpcCall(ip, ip)))//同步转换为异步
.collect(Collectors.toList());//收集结果
//3.等待所有异步任务执行完毕
List<String> resultList = futureList.stream()
.map(future -> future.join())
//同步等待结果
.collect(Collectors.toList());
//对结果进行收集
// 4.输出
resultList.stream().forEach(r -> System.out.println(r));
System.out.println("cost:" + (System.currentTimeMillis() - start));
注意:具体这10个rpc请求是否全部并发运行取决于CompletableFuture内线程池内线程的个数,如果你的机器是单核的或者线程池内线程个数为1,那么这10个任务还是会顺序执行的。
我们了解了CompletableFuture如何解决其缺点,以及CompletableFuture与JDK Stream是如何完美结合的,可知使用CompletableFuture实现异步编程属于声明式编程,一般情况下不需要我们显式创建线程池并提交任务到线程池,这大大减轻了编程者的负担。