[转载]Java StreamAPI 详解

//Java8 中的 Collection 接口被扩展，提供两个获取流的方法 :
default Stream<E> stream() //: 返回一个顺序流
default Stream<E> parallelStream()// : 返回一个并行流

// Collection 提供了两个方法  stream() 与 parallelStream()
List<String> list = new ArrayList<>();
Stream<String> stream = list.stream(); // 获取一个顺序流
Stream<String> parallelStream = list.parallelStream(); // 获取一个并行流

//由数组创建流
//Java8 中的 Arrays 的静态方法 stream() 可以获取数组流 :
static <T> Stream<T> stream(T[] array) : //返回一个流
//重载形式，能够处理对应基本类型的数组 :
public static IntStream stream(int[] array)
public static LongStream stream(long[] array)
public static DoubleStream stream(double[] array)

// 通过 Arrays 中的 stream() 获取一个数组流
Stream<Integer> stream1 = Arrays.stream(new Integer[10]);

//由值创建流
//可以使用静态方法 Stream.of(), 通过显示值创建一个流，它可以接收任意数量的参数
public static<T> Stream<T> of(T... values) : 返回一个流

// 通过 Stream 类中静态方法 of()
Stream<Integer> stream2 = Stream.of(1, 2, 3, 4, 5, 6);
//注 : Stream.of静态方法 底层就是 Arrays.stream 静态方法

//由函数创建流 : 创建无限流
//可以使用静态方法 Stream.iterate() 和 Stream.generate(), 创建无限流迭代
public static<T> Stream<T> iterate(final T seed, final UnaryOperator<T> f)
//生成
public static<T> Stream<T> generate(Supplier<T> s)

// 创建无限流
// 迭代
Stream<Integer> stream3 = Stream.iterate(0, (x) -> x + 2).limit(10);
stream3.forEach(System.out::println);

// 生成
Stream<Double> stream4 = Stream.generate(Math::random).limit(2);
stream4.forEach(System.out::println);

Iterator<Employee> it = emps.iterator();
while (it.hasNext()) {
    System.out.println(it.next());
}
emps.stream().filter(e -> e.getSalary() >= 5000).limit(3).forEach(System.out::println);
emps.parallelStream().filter((e) -> e.getSalary() >= 5000).skip(2).forEach(System.out::println);
emps.stream().distinct().forEach(System.out::println);

Stream<String> str = emps.stream().map(Employee::getName);

List<String> strList = Arrays.asList("aaa", "bbb", "ccc", "ddd", "eee");
Stream<String> stream1 = strList.stream().map(String::toUpperCase);
stream1.forEach(System.out::println);

public static Stream<Character> filterCharacter(String str) {
    List<Character> list = new ArrayList<>();
    for (Character ch : str.toCharArray()) {
        list.add(ch);
    }
    return list.stream();
}

Stream<Stream<Character>> stream2 = strList.stream().map(TestStreamaAPI2::filterCharacter);
stream2.forEach(sm -> sm.forEach(System.out::println));

Stream<Character> stream3 = strList.stream().flatMap(TestStreamaAPI2::filterCharacter);
stream3.forEach(System.out::println);

Integer[] nums = new Integer[]{1, 2, 3, 4, 5};
Arrays.stream(nums).map((x) -> x * x).forEach(System.out::println);

emps.stream().map(Employee::getName).sorted().forEach(System.out::println);
emps.stream().sorted((x, y) -> {
            if (x.getAge() == y.getAge()) {
                return x.getName().compareTo(y.getName());
            } else {
                return Integer.compare(x.getAge(), y.getAge());
            }
        }).forEach(System.out::println);

boolean bl = emps.stream().allMatch(e -> e.getStatus().equals(Status.BUSY));
System.out.println(bl);

boolean bl1 = emps.stream().anyMatch(e -> e.getStatus().equals(Status.BUSY));
System.out.println(bl1);

boolean bl2 = emps.stream().noneMatch(e -> e.getStatus().equals(Status.BUSY));
System.out.println(bl2);

Optional<Employee> op1 = emps.stream().sorted((e1, e2) -> Double.compare(e1.getSalary(), e2.getSalary())).findFirst();
System.out.println(op1.get());

Optional<Employee> op2 = emps.parallelStream().filter((e) -> e.getStatus().equals(Status.FREE)).findAny();
System.out.println(op2.get());

long count = emps.stream().filter((e) -> e.getStatus().equals(Status.FREE)).count();
System.out.println(count);

Optional<Double> op = emps.stream().map(Employee::getSalary).max(Double::compare);
System.out.println(op.get());

Optional<Employee> op2 = emps.stream().min((e1, e2) -> Double.compare(e1.getSalary(), e2.getSalary()));
System.out.println(op2.get());

//注 : 流进行了终止操作后，不能再次使用
Stream<Employee> stream = emps.stream().filter(e -> e.getStatus().equals(Status.FREE));
long count = stream.count();
System.out.println(count);

stream.map(Employee::getSalary).max(Double::compare);

//注 : map 和 reduce 的连接通常称为 map-reduce 模式，因 Google 用它来进行网络搜索而出名
List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
Integer sum = list.stream().reduce(0, (x, y) -> x + y);
System.out.println(sum);

Optional<Double> op = emps.stream().map(Employee::getSalary).reduce(Double::sum);
System.out.println(op.get());

//需求 : 搜索名字中 “六” 出现的次数
Optional<Integer> sum = emps.stream()
        .map(Employee::getName)
        .flatMap(TestStreamaAPI2::filterCharacter)
        .map((ch) -> {
            if (ch.equals(‘六‘)) {
                return 1;
            } else {
                return 0;
            }
        }).reduce(Integer::sum);
System.out.println(sum.get());

List<String> list = emps.stream().map(Employee::getName).collect(Collectors.toList());
list.forEach(System.out::println);

Set<String> set = emps.stream().map(Employee::getName).collect(Collectors.toSet());
set.forEach(System.out::println);

HashSet<String> hs = emps.stream().map(Employee::getName).collect(Collectors.toCollection(HashSet::new));
hs.forEach(System.out::println);

Optional<Double> max = emps.stream().map(Employee::getSalary).collect(Collectors.maxBy(Double::compare));
System.out.println(max.get());

Optional<Employee> op = emps.stream().collect(Collectors.minBy((e1, e2) -> Double.compare(e1.getSalary(), e2.getSalary())));
System.out.println(op.get());

Double sum = emps.stream().collect(Collectors.summingDouble(Employee::getSalary));
System.out.println(sum);

Double avg = emps.stream().collect(Collectors.averagingDouble(Employee::getSalary));
System.out.println(avg);

Long count = emps.stream().collect(Collectors.counting());
System.out.println(count);

DoubleSummaryStatistics dss = emps.stream().collect(Collectors.summarizingDouble(Employee::getSalary));
System.out.println(dss.getMax());

// 分组

Map<Status, List<Employee>> map = emps.stream().collect(Collectors.groupingBy(Employee::getStatus));
System.out.println(map);

// 多级分组
Map<Status, Map<String, List<Employee>>> map = emps.stream()
        .collect(Collectors.groupingBy(Employee::getStatus, Collectors.groupingBy((e) -> {
            if (e.getAge() >= 60) {
                return "老年";
            } else if (e.getAge() >= 35) {
                return "中年";
            } else {
                return "成年";
            }
        })));
System.out.println(map);

String str = emps.stream().map(Employee::getName)
        .collect(Collectors.joining(",", "----", "----"));
System.out.println(str);
Optional<Double> sum = emps.stream().map(Employee::getSalary).collect(Collectors.reducing(Double::sum));
System.out.println(sum.get());

//交易员类
public class Trader {
    private String name;
    private String city;
    public Trader() {
    }
    // ...
    @Override
    public String toString() {
        return "Trader [name=" + name + ", city=" + city + "]";
    }
}

//交易类
public class Transaction {
    private Trader trader;
    private int year;
    private int value;
    public Transaction() {
    }
    public Transaction(Trader trader, int year, int value) {
        this.trader = trader;
        this.year = year;
        this.value = value;
    }
    // ...
    @Override
    public String toString() {
        return "Transaction [trader=" + trader + ", year=" + year + ", value=" + value + "]";
    }
}

//初始化数据
List<Transaction> transactions = null;

@Before
public void before() {
    Trader raoul = new Trader("Raoul", "Cambridge");
    Trader mario = new Trader("Mario", "Milan");
    Trader alan = new Trader("Alan", "Cambridge");
    Trader brian = new Trader("Brian", "Cambridge");

    transactions = Arrays.asList(
            new Transaction(brian, 2011, 300),
            new Transaction(raoul, 2012, 1000),
            new Transaction(raoul, 2011, 400),
            new Transaction(mario, 2012, 710),
            new Transaction(mario, 2012, 700),
            new Transaction(alan, 2012, 950)
    );
}

//1. 找出2011年发生的所有交易， 并按交易额排序(从低到高)
transactions.stream().filter((t) -> t.getYear() == 2011).sorted((t1, t2) -> Integer.compare(t1.getValue(), t2.getValue())).forEach(System.out::println);

//2. 交易员都在哪些不同的城市工作过
transactions.stream().map((t) -> t.getTrader().getCity()).distinct().forEach(System.out::println);

//3. 查找所有来自剑桥的交易员，并按姓名排序
transactions.stream().filter((t) -> t.getTrader().getCity().equals("Cambridge")).map(Transaction::getTrader)
        .sorted(Comparator.comparing(Trader::getName)).distinct().forEach(System.out::println);

//4. 返回所有交易员的姓名字符串，按字母顺序排序
transactions.stream().map((t) -> t.getTrader().getName()).sorted().forEach(System.out::println);

String str = transactions.stream().map((t) -> t.getTrader().getName()).sorted().reduce("", String::concat);
System.out.println(str);

public static Stream<String> filterCharacter(String str) {
    List<String> list = new ArrayList<>();
    for (Character ch : str.toCharArray()) {
        list.add(ch.toString());
    }
    return list.stream();
}
transactions.stream().map((t) -> t.getTrader().getName()).flatMap(TestTransaction::filterCharacter).sorted(String::compareToIgnoreCase).forEach(System.out::print);

//5. 有没有交易员是在米兰工作的
boolean bl = transactions.stream().anyMatch((t) -> t.getTrader().getCity().equals("Milan"));
System.out.println(bl);

//6. 打印生活在剑桥的交易员的所有交易额
Optional<Integer> sum = transactions.stream().filter((e) -> e.getTrader().getCity().equals("Cambridge"))
        .map(Transaction::getValue).reduce(Integer::sum);
System.out.println(sum.get());

//7. 所有交易中，最高的交易额是多少
Optional<Integer> max = transactions.stream().map(Transaction::getValue).max(Integer::compare);
System.out.println(max.get());

//8. 找到交易额最小的交易
Optional<Transaction> op = transactions.stream().min((t1, t2) -> Integer.compare(t1.getValue(), t2.getValue()));
System.out.println(op.get());

long start = System.currentTimeMillis();

long sum = 0L;
for (long i = 0L; i <= 10000000000L; i++) {
    sum += i;
}
System.out.println(sum);

long end = System.currentTimeMillis();
System.out.println("耗费的时间为: " + (end - start)); //34-3174-3132-4227-4223-31583

public class ForkJoinCalculate extends RecursiveTask<Long> {
    private static final long serialVersionUID = 13475679780L;
    private long start;
    private long end;
    private static final long THRESHOLD = 10000L; //临界值
    public ForkJoinCalculate(long start, long end) {
        this.start = start;
        this.end = end;
    }
    @Override
    protected Long compute() {
        long length = end - start;
        if (length <= THRESHOLD) {
            long sum = 0;
            for (long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        } else {
            long middle = (start + end) / 2;
            ForkJoinCalculate left = new ForkJoinCalculate(start, middle);
            left.fork(); //拆分，并将该子任务压入线程队列

            ForkJoinCalculate right = new ForkJoinCalculate(middle + 1, end);
            right.fork();
            return left.join() + right.join();
        }
    }
}

long start = System.currentTimeMillis();

ForkJoinPool pool = new ForkJoinPool();
ForkJoinTask<Long> task = new ForkJoinCalculate(0L, 10000000000L);

long sum = pool.invoke(task);
System.out.println(sum);

long end = System.currentTimeMillis();
System.out.println("耗费的时间为: " + (end - start)); //112-1953-1988-2654-2647-20663-113808

long start = System.currentTimeMillis();

Long sum = LongStream.rangeClosed(0L, 10000000000L).parallel().sum();
System.out.println(sum);

long end = System.currentTimeMillis();
System.out.println("耗费的时间为: " + (end - start)); //2061-2053-2086-18926