java工程师要会的大数据之Stream流

原创

Joseph_青椒

发布于 2023-08-09 19:53:39

6690

发布于 2023-08-09 19:53:39

文章被收录于专栏：java_josephjava_joseph

文章初衷：用Flink做大数据方面的业务，用到了函数式编程，这里stream忘的很多，平时用的也比较少，所以这里作为大数据专栏的第一个，第一遍接触函数式编程的，可能很不舒服，我也是，于是乎就静下心，写了这篇文章，看不懂的多看几遍，其实很简单！所以这篇文章可能是crud，大家见谅哈，有对应基础的可以跳过，直接看俺后面的教程

认识函数式编程

什么式函数式编程，什么是Stream流

在JDK8之前，Java是不⽀持函数式编程的，所谓的函数编程，即可理解是将⼀个函数（也称为“⾏为”）作为⼀个参数进⾏传递，⾯向对象编程是对数据的抽象（各种各样的POJO类），⽽函数式编程则是对⾏为的抽象（将⾏为作为⼀个参数进⾏传递

我当时看这里很懵，直接理解为代码中，把对象换成一个操作的过程

jdk8之前创建线程

new Thread(new Runnable() {
 @Override
 public void run() {
    System.out.println("Java架构教程");
 }
 })；

stream流这样做

new Thread(()-> System.out.println("hhhh")).start();

再看一个例子

集合容器⾥⾯的字符串排序使⽤前

List<String> list = Arrays.asList("aaa","bbb","ggg","ccc");
        Collections.sort(list, new Comparator<String>() {
            @Override
            public int compare(String a, String b) {
                return b.compareTo(a);
            }
        });
        for(String str : list){
            System.out.println(str);
        }

使用后

Collections.sort(list,(a,b)->b.compareTo(a));
        for(String str : list){
            System.out.println(str);
        }

lambda表达式使⽤场景(前提)：⼀个接⼝中只包含⼀个⽅法，则可以使⽤Lambda表达式，这样的接⼝称之为“函数接⼝” 语法： (params) -> expression

比如上面的compare方法，俩参数，写的时候直接(a,b)

第⼀部分为括号内⽤逗号分隔的形式参数，参数是函数式接⼝⾥⾯⽅法的参数；

第⼆部分为⼀个箭头符号：->；

第三部分为⽅法体，可以是表达式和代码块参数列表：

括号中参数列表的数据类型可以省略不写括号中的参数只有⼀个，那么参数类型和()都可以省略不写⽅法体：

如果{}中的代码只有⼀⾏，可以省略{}，return，分号，要⼀起省略，其他则需要加上

也就是说，一行，可以省略{}、；、return。多行的话，；、return都不能省略

其实就是对匿名内部类方式的重构，提高开发效率

自定义函数式编程

为了更好的理解lamada表达式，这里自定义一个函数式编程

上面讲到，函数式编程其实就是对匿名内部类的重构，前提是一个接口只包含一个方法

1、自定义接口，加@FunctionalInterface注解，这个注解能保证这个接口只有一个方法

@FunctionalInterface
public interface OperFunction<R,T> {
//R是返回值，T是参数
    R operator(T t1,T t2);

}

编写⼀个⽅法，输⼊需要操做的数据和接⼝，这里就是定义好函数

 public static Integer operator(Integer x,Integer y,OperFunction<Integer,Integer> of){
        return of.operator(x,y);
    }

在调⽤⽅法时传⼊数据和 lambda 表达式，⽤来操作数据,就是传入具体的过程

System.out.println(operator(10,20,(x,y)->x+y));
System.out.println(operator(10,20,(x,y)->x/y));

上面是更解耦的方法，定义一个就可以实现+-*/，常规这样更直白。

就是把接口实例，接受一个函数，然后执行接口的方法，就会调用这个函数，其实很好理解的

简单理解的话，就是把对象换成函数行为，执行接口的方法，就会执行这个函数，这就是函数式编程

OperFunction<Integer,Integer> function = (x,y)->x+y;
        System.out.println(function.operator(10,1));

函数式编程Function

工作中式不用我们自己定义Function的

jdk帮忙做了这些Function

Consumer<T> //消费型接口，有入参，无返回值

void accept(T t);

Supplier : 供给型接⼝：⽆⼊参，有返回值

T get();

Function : 函数型接⼝：有⼊参，有返回值

R apply(T t);

Predicate : 断⾔型接⼝：有⼊参，有返回值

返回值类型确定是boolean boolean test(T t)

Function函数型接⼝

核心就是把对象换成行为，然后apply就是执行这个行为

   Function<Integer,Integer> function = p ->p*10;
        System.out.println(function.apply(10));

BiFunction

Function只有一个入参，一个返回值，

两个入参两个，一个返回值是BiFunction

@FunctionalInterface

public interface BiFunction {

R apply(T t, U u);

}

玩儿个四则运算来练习一下（自定义函数编程差不多）

再说下思想哈，就是让接口接受一个函数行为，然后调用

public class Main {
    public static void main(String[] args) {

        System.out.println(operator(10,20,(x,y)->x+y));
        System.out.println(operator(10,20,(x,y)->x-y));
        System.out.println(operator(10,20,(x,y)->x*y));

    }

    /**
     * z这个方法只是简单解耦了一下，可以让在调用时写具体函数行为，不用重复定义BiFunction
     * @param x
     * @param y
     * @param bf
     * @return
     */
    public static Integer operator(Integer x, Integer y, BiFunction<Integer,Integer,Integer> bf){
        return bf.apply(x,y);
    }
}

看到这里的应该能看懂吧？？

看不懂，看下面的，不解耦的写法

public class Main {
    public static void main(String[] args) {

       BiFunction<Integer,Integer,Integer>  func = (x,y)->x+y;
        System.out.println(func.apply(1,2));

    }


}

Consumer消费型接口

消费指的是这个接口只进不出，适用于发送日志、短信这样的业务

public class Main {
    public static void main(String[] args) {

        Consumer<String> consumer = obj->{
            System.out.println(obj);
            System.out.println("调用短信接口发送短信");
        };
        sendMsg("1254654",consumer);

    }
    public static void sendMsg(String phone,Consumer<String> consumer){
        consumer.accept(phone);
    }


}
===============写法二==============
public class Main {
    public static void main(String[] args) {
        
        sendMsg("1254654",obj->{
            System.out.println(obj);
            System.out.println("调用短信接口发送短信");
        });

    }
    public static void sendMsg(String phone,Consumer<String> consumer){
        consumer.accept(phone);
    }


}

jdk8，遍历结合的forEach

public class Main {
    public static void main(String[] args) {

        List<Integer>  list  = Arrays.asList(1,2,3,4);
        list.forEach(obj-> System.out.println(obj));

    }

}

    * }</pre>他就是用Consummer实现的！！！
     *
     * @param action The action to be performed for each element
     * @throws NullPointerException if the specified action is null
     * @since 1.8
     */
    default void forEach(Consumer<? super T> action) {
        Objects.requireNonNull(action);
        for (T t : this) {
            action.accept(t);
        }
    }

可以看到，这个forEach方法编写的时候，就是调用函数式编程的执行方法，用户使用的时候传入具体函数行为，帮我们执行

Supplier供应型接口

适用于工厂模式，不需要提供参数的,这个不太常用奥，主要两外三个

@FunctionalInterface
public interface Supplier<T> {

    /**
     * Gets a result.
     *
     * @return a result
     */
    T get();
}

public class Main {
    public static void main(String[] args) {

        Student student = newStudent()
    }
    public static Student newStudent(){
        Supplier<Student> supplier = ()->{
            Student student = new Student();
            student.setName("默认名称");
            return student;
        };
        return supplier.get();
    }

}
class Student{
    String name;
    int age;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }
}

Predicate: 断⾔型接⼝

断言，就是看是否符合，符合就是true，不符合为false

public class Main {
    public static void main(String[] args) {

        List<String> list = Arrays.asList("awewrwe","vdssdsd","aoooo","psdddsd");
        List<String> results = fliter(list,obj->obj.startsWith("a"));
        System.out.println(list);
        System.out.println(results);
    }
    public static List<String> fliter(List<String> list, Predicate<String> predicate){
        List<String> results = new ArrayList<>();
        for(String str : list){
            //test也就是断言的逻辑是要我们输入函数行为的
            if(predicate.test(str)){
                results.add(str);
            }
        }
        return results;
    }


}

stream.fliter(),就是采用的Prediacate断言型接口

总结：

到这里，就算基本是掌握函数式编程了，无非就是用接口接受一个函数行为，而不是对象，执行接口响应的方法，就会执行函数行为，jdk封装的，比如stream流，就是把函数接口定义好，用户调用的时候传入具体的函数行为，就会执行，如stream的Fliter，就是采用的Predicate断言型接口。程序员就写入断言的逻辑，就可以了。

方法与构造函数引用

新的调用方法，lamada表达式，可用::双冒号来调用方法或者构造方法，这种方式叫做方法引用

语法：

左边式容器（类名、实例名）中间式"::" 右面式相应的方法名

静态⽅法，则是ClassName::methodName。如 Object ::equals

实例⽅法，则是Instance::methodName

构造函数，则是类名::new;

看懂下面的demo即可

public class Main {
    public static void main(String[] args) {
        //使用双冒号::来构造静态函数的引用
        Function<String,Integer> fun= Integer::parseInt;
        Integer value = fun.apply("1027");
        System.out.println(value);
        //使用双冒号：：构造非静态函数引用
        String content = "hello";
        Function<Integer,String> func = content::substring;
        String result = func.apply(1);
        System.out.println(result);
        //使用::构造  构造函数引用 根据Function的参数类型，选择执行哪个构造函数
        BiFunction<Integer,String,User> biFunction = User::new;
        User user = biFunction.apply(2, "哈哈哈");
        System.out.println(user);
        //函数引用，将函数当作参数,更好的解耦
        sayHello(String::toUpperCase,"joseph,java");

    }
    private static void sayHello(Function<String,String> func,String param){
        String result = func.apply(param);
        System.out.println(result);
    }


}
class User{
    private String name;
    private int age;
    public User(){

    }
    public User(int age,String name){
        this.age = age;
        this.name=  name;
    }
    public User(String name){
        this.name=  name;
    }
    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "User{" +
                "name='" + name + '\'' +
                ", age=" + age +
                '}';
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }
}

Stream流

stream流，可对集合中的元素进行串行or并行的流水线操作

操作详情数据元素便是原始集合，如List、Set、Map等

⽣成流，可以是串⾏流stream() 或者并⾏流 parallelStream()

中间操作，可以是排序，聚合，过滤，转换等终端操作，很多流操作本身就会返回⼀个流，所以多个操作可以直接连接起来，最后统⼀进⾏收集

比如上面讲的fliter。就是Stream.java中的，集合框架中有stream方法，

快速认识一下

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("哈哈哈", "嘿嘿嘿", "嘤嘤嘤");
        //给每个元素拼个文字
        List<String> resultList = list.stream().map(obj -> "我在" + obj).collect(Collectors.toList());
        System.out.println(resultList);

    }
}

map

讲流中的每个元素做映射，

场景，比如项目中的DO转VO，

public class Main {
    public static void main(String[] args) {
        List<UserDO> userDOList = Arrays.asList(new UserDO(2, "小明", "adg"), new UserDO(3, "小d", "adgss"), new UserDO(2, "小xi", "adga"));
        //现在要返回前端VO，没有pwd，用stream流处理
        List<UserVO> userVOList = userDOList.stream().map(obj -> {
            UserVO userVO = new UserVO(obj.getName(), obj.getAge());
            return userVO;
        }).collect(Collectors.toList());
        System.out.println(userDOList);
        System.out.println(userVOList);
    }
}
class UserDO {
    private String name;
    private int age;
    private String pwd;
    public UserDO(){

    }
    public UserDO(int age, String name,String pwd){
        this.age = age;
        this.name=  name;
        this.pwd = pwd;
    }
    public String getPwd() {
        return pwd;
    }

    public void setPwd(String pwd) {
        this.pwd = pwd;
    }



    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "UserDO{" +
                "name='" + name + '\'' +
                ", age=" + age +
                ", pwd='" + pwd + '\'' +
                '}';
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }
}
class UserVO{
    private String name;
    private int age;
    public UserVO(){

    }
    public UserVO(String name,int age){
        this.name = name;
        this.age = age;
    }
    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    @Override
    public String toString() {
        return "UserVO{" +
                "name='" + name + '\'' +
                ", age=" + age +
                '}';
    }
}

fliter

fliter就是过滤器，之前讲过，那是我们手写了一个过滤器，用的就是Predicate断言式接口

stream直接写好了，咱直接用

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("aaaaaaa", "bb");
        //现在要返回前端VO，没有pwd，用stream流处理
        List<String> list1 = list.stream().filter(obj -> obj.length() > 5).collect(Collectors.toList());
        System.out.println(list1);
    }
}

sorted

排序

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("cc", "bb","aa");
        //现在要返回前端VO，没有pwd，用stream流处理
        List<String> list1 = list.stream().sorted().collect(Collectors.toList());
        System.out.println(list1);
    }
}

默认式升序，也可以

Comparator提供了函数式编程使用的comparing

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("cc", "bbaaa","aaa");
        //现在要返回前端VO，没有pwd，用stream流处理
        List<String> list1 = list.stream().sorted(Comparator.comparing(obj->obj.length())).collect(Collectors.toList());
        System.out.println(list1);
        //上面是根据长度升序，也可以降序
        List<String> list2 = list.stream().sorted(Comparator.comparing(obj->obj.length())).collect(Collectors.toList());
        System.out.println(list2);
        //写法2  方法引用
        List<String> list3 = list.stream().sorted(Comparator.comparing(String::length).reversed()).collect(Collectors.toList());
        System.out.println(list3);

    }
}

limit

截取指定条数

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("cc", "bbaaa","aaa");

        List<String> list3 = list.stream().sorted(Comparator.comparing(String::length).reversed()).limit(2).collect(Collectors.toList());
        System.out.println(list3);

    }
}

allMatch函数

检查是否匹配所有元素，只有全部符合才返回true

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("springboot", "springcloud", "redis", "git", "netty", "java", "html", "docker");
        boolean flag = list.stream().allMatch(obj->obj.length()>1);
        System.out.println(flag);
    }
}

anyMatch函数

检查是否⾄少匹配⼀个元素

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("springboot", "sprissssssssngclosssssud", "redis", "git", "netty", "java", "html", "docker");
        boolean flag = list.stream().anyMatch(obj->obj.length()>18);
        System.out.println(flag);
    }
}

max

最大值

Optional<T> max(Comparator<? super T> comparator);
看到。需要我们传入一个比较器，而comparator是支持lamada函数编程的

public class Main {
    public static void main(String[] args) {
        
        List<String> list = Arrays.asList("springboot", "sprissssssssngclosssssud", "redis", "git", "netty", "java", "html", "docker");
        Optional<String> max = list.stream().max((s1, s2) -> {
            return Integer.compare(s1.length(), s2.length());
        });
        String maxString = max.get();
        System.out.println(maxString);
    }
}//写法2
 Optional<String> max = list.stream().max(Comparator.comparingInt(String::length));

min这里就不讲了，就是取反，这个我们自定义排序规则的话，也不太需要这个

并行流parallelStream

提高集合做重复操作的效率，通过fork/join框架维护线程池，可以分割任务，父任务拆分成子任务，

穿行流与并行流区别

public class Main {
    public static void main(String[] args) {
        List<Integer> integers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9);
//        integers.stream().forEach(System.out::println);
        integers.parallelStream().forEach(System.out::println);

    }
}

首先并行流会乱序，

其次，paralleStream并行不一定比Stream串性快

数据量小的时候，串行会更快，ForkJoin会耗性能

多数情况下，并行比串行快，但是需要注意线程安全问题，要使用线程安全的集合

public class Main {
    public static void main(String[] args) {
        for (int i = 0; i < 10; i++) {
            //这里会有线程安全的问题，比如99的时候，两个线程都判断到99.就会越界
//            List list = new ArrayList();
            CopyOnWriteArrayList<Object> list = new CopyOnWriteArrayList<>();
            IntStream.range(0,100).parallel().forEach(list::add);
            System.out.println(list.size());
        }
    }
}

jvm一般不会创建太多，推荐使用stream串行流

reduce聚合

根据⼀定的规则将Stream中的元素进⾏计算后返回⼀个唯⼀的值

Optional reduce(BinaryOperator accumulator);

public class Main {
    public static void main(String[] args) {
        int value = Stream.of(1, 2, 3, 4, 5).reduce((item1, item2) -> item1 + item2).get();
        System.out.println(value);
        //不用lamada写法
        Integer value2 = Stream.of(1, 2, 3, 4, 5).reduce(new BinaryOperator<Integer>() {
            @Override
            public Integer apply(Integer integer, Integer integer2) {
                return integer + integer2;
            }
        }).get();
        System.out.println(value2);
        //用法2,来一个初始值
        Stream.of(1, 2, 3, 4, 5).reduce(100,(item1,item2)->item1+item2);

    }
}

foreach

上面讲过一次，是通过Consummer实现的，

public class Main {
    public static void main(String[] args) {
        List<Integer> integers = Arrays.asList(1, 2, 3, 41, 2, 3, 1, 1, 2);
        integers.forEach(obj-> System.out.println(obj));
        integers.forEach(System.out::println);
    }
}

需要注意的是：循环里不能操作外部变量，且continue和break不能用，和普通的for循环，增强for循环这些是不一样的，return也是没有意义的

收集器collect与集合统计

上面讲完stream，一些基本使用，里面用到了collect，就是流处理的终端操作，收集处理完的数据

//重载⽅法⼀
 <R> R collect(Supplier<R> supplier, BiConsumer<R, ? super T>
accumulator, BiConsumer<R, R>combiner);
 //重载⽅法⼆
 <R, A> R collect(Collector<? super T, A, R> collector)

一般不用自定义，用他们做好的就行，重载方法2

传入Collector就可，jdk封装的够我们用了

比如Collector.toList

public static <T>
    Collector<T, ?, List<T>> toList() {
        return new CollectorImpl<>((Supplier<List<T>>) ArrayList::new, List::add,
                                   (left, right) -> { left.addAll(right); return left; },
                                   CH_ID);
    }

Collectors.toCollection(LinkedList::new)//这里直接创建自定义的，传入构造函数引用即可

Collectots.toCollection(TreeSet::new)

也可以toSet什么的，都封装好了，咱直接用就好了！

joinging

是Collector的一个实现 CollectorImpl

   * {@code String}, in encounter order
     */
    public static Collector<CharSequence, ?, String> joining() {
        return new CollectorImpl<CharSequence, StringBuilder, String>(
                StringBuilder::new, StringBuilder::append,
                (r1, r2) -> { r1.append(r2); return r1; },
                StringBuilder::toString, CH_NOID);
    }

public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("嘿嘿","哈哈");
        String result1 = list.stream().collect(Collectors.joining());
        System.out.println(result1);
        String result2 = list.stream().collect(Collectors.joining("||"));
        System.out.println(result2);
        String result3 = list.stream().collect(Collectors.joining("||","前","后"));
        System.out.println(result3);
    }
}

partitioningBy分组

public static <T>
    Collector<T, ?, Map<Boolean, List<T>>> partitioningBy(Predicate<? super T> predicate) {
        return partitioningBy(predicate, toList());
    }

来个例子：根据list⾥⾯进⾏分组，字符串⻓度⼤于4的为⼀组，其他为另外⼀组

import static java.util.stream.Collectors.partitioningBy;
public class Main {
    public static void main(String[] args) {
        List<String> list = Arrays.asList("java","springboot","HTML5","CSS3");
        Map<Boolean, List<String>> res = list.stream().collect(partitioningBy(obj -> obj.length() > 4));
        System.out.println(res);
    }
}

groupBy分组

public class Main {
    public static void main(String[] args) {
        List<Student> students = Arrays.asList(new Student("⼴东", 23), new
                Student("⼴东", 24), new Student("⼴东", 23),new Student("北京", 22), new
                Student("北京", 20), new Student("北京", 20),new Student("海南", 25));
        //根据学生所在省份进行分组
        Map<String, List<Student>> res = students.stream().collect(Collectors.groupingBy(obj -> obj.getProvice()));
        System.out.println(res);
        res.forEach((key,value)->{
            System.out.println("========");
            System.out.println(key);
            value.forEach(obj->{
                System.out.println(obj.getAge());
            });
        });
    }
}
class Student{
    private int age;
    private String provice;
    public Student(){

    }
    public Student( String provice,int age) {
        this.age = age;
        this.provice = provice;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public String getProvice() {
        return provice;
    }

    public void setProvice(String provice) {
        this.provice = provice;
    }

    @Override
    public String toString() {
        return "Student{" +
                "age=" + age +
                ", provice='" + provice + '\'' +
                '}';
    }
}

分组之后，还要统计groupBy(重载)

groupby还有一个重载函数

public class Main {
    public static void main(String[] args) {
        List<Student> students = Arrays.asList(new Student("⼴东", 23), new
                Student("⼴东", 24), new Student("⼴东", 23),new Student("北京", 22), new
                Student("北京", 20), new Student("北京", 20),new Student("海南", 25));
        //根据学生所在省份进行分组
        Map<String, Long> map = students.stream().collect(Collectors.groupingBy(obj -> obj.getProvice(), Collectors.counting()));
        System.out.println(map);
        map.forEach((key,value)->{
            System.out.println(key+"省有："+value);
        });
    }
}

summarizing集合统计

summarizingInt

summarizingLong

summarizingDouble

public class Main {
    public static void main(String[] args) {
        List<Student> students = Arrays.asList(new Student("⼴东", 23), new
                Student("⼴东", 24), new Student("⼴东", 23),new Student("北京", 22), new
                Student("北京", 20), new Student("北京", 20),new Student("海南", 25));
       // ：统计学⽣的各个年龄信息
        IntSummaryStatistics summaryStatistics = students.stream().collect(Collectors.summarizingInt(Student::getAge));
        System.out.println("平均值"+summaryStatistics.getAverage());
        System.out.println("人数"+summaryStatistics.getCount());
        System.out.println("最大值"+summaryStatistics.getMax());
        System.out.println("最小值"+summaryStatistics.getMin());
        System.out.println("年龄总和"+summaryStatistics.getSum());
    }
}
class Student{
    private int age;
    private String provice;
    public Student(){

    }
    public Student( String provice,int age) {
        this.age = age;
        this.provice = provice;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }

    public String getProvice() {
        return provice;
    }

    public void setProvice(String provice) {
        this.provice = provice;
    }

    @Override
    public String toString() {
        return "Student{" +
                "age=" + age +
                ", provice='" + provice + '\'' +
                '}';
    }
}

结束了！，大家看完这些demo，自己debug一下，肯定能看懂！

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

大数据

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

大数据

登录后参与评论

0 条评论

热度