怎樣用Java 8優雅的開發業務

怎樣用Java 8優雅的開發業務
- 函式式編程
  - 流式編程
    - 基本原理
    - 案例
- 優雅的空處理
- 新的并發工具類CompletableFuture
  - 單機批處理多執行緒執行模型
    - 模型
    - 案例
  - 調整執行緒池的大小
  - 并行——使用流還是CompletableFutures？
- 日期和時間API
- 專案地址
- 參考

函式式編程

匿名函式

λ演算

流式編程

基本原理

在Java中流式編程的基本原理有兩點，

構建流
資料流轉(流水線)
規約

IntStream.rangeClosed(1, 100) // 1. 構建流
    .mapToObj(String::valueOf)// 2. 資料流轉(流水線)
    .collect(joining());      // 3. 規約

案例

英雄的主位置一共有幾類，分別是什么

@Test
fun t1() {
    // 英雄的主位置一共有幾類，分別是什么
    // 映射
    val roleMains = heroes.map(Hero::getRoleMain)
        // 過濾為空的資料
        .filter(Objects::nonNull)
        // 去重
        .distinct()
    println(roleMains.size)
    println(roleMains)
}

@Test
public void t1() {
    // 英雄的主位置一共有幾類，分別是什么
    List<String> roleMains = heroes.stream()
            // 映射
            .map(Hero::getRoleMain)
            // 過濾為空的資料
            .filter(Objects::nonNull)
            // 去重
            .distinct()
            // 收集串列
            .collect(toList());
    System.out.println(roleMains.size());
    System.out.println(roleMains);
}

英雄按主次位置分組后，輸出每個分組有多少英雄，其中：近戰英雄有多少位，遠程英雄有多少位

@Test
fun t2() {
    // 英雄按主次位置分組后，輸出每個分組有多少英雄，其中：近戰英雄有多少位，遠程英雄有多少位

    // 主次位置分組的英雄數量
    val groupHeroCount = heroes.groupingBy {
        Pair.of(it.roleMain, it.roleAssist)
    }.eachCount()

    // 主次分組后，再按攻擊范圍分組的英雄數量
    val groupThenGroupCount = heroes.groupBy {
        Pair.of(it.roleMain, it.roleAssist)
    }.map {
        val value = https://www.cnblogs.com/switchvov/p/it.value.groupingBy(Hero::getAttackRange).eachCount()
        Pair.of(it.key, value)
    }.associateBy({ it.left }, { it.value })

    // 遍歷輸出
    groupThenGroupCount.forEach { (groupKey, groupValue) ->
        val groupingCount = groupHeroCount[groupKey]
        print("英雄分組key為:$groupKey;英雄數量:$groupingCount;")
        groupValue.forEach { (countKey, countValue) ->
            print("英雄攻擊范圍:$countKey;英雄數量:$countValue;")
        }
        println()
    }
}

@Test
public void t2() {
    // 英雄按主次位置分組后，輸出每個分組有多少英雄，其中：近戰英雄有多少位，遠程英雄有多少位

    // 主次位置分組的英雄數量
    Map<Pair<String, String>, Long> groupHeroCount = heroes.stream()
            .collect(groupingBy(hero -> Pair.of(hero.getRoleMain(), hero.getRoleAssist()), counting()));

    // 主次分組后，再按攻擊范圍分組的英雄數量
    Map<Pair<String, String>, Map<String, Long>> groupThenGroupCount = heroes.stream()
            .collect(groupingBy(hero -> Pair.of(hero.getRoleMain(), hero.getRoleAssist()),
                    groupingBy(Hero::getAttackRange, counting())));

    // 遍歷輸出
    groupThenGroupCount.forEach((groupKey, groupValue) -> {
        Long groupingCount = groupHeroCount.get(groupKey);
        System.out.print("英雄分組key為:" + groupKey + ";英雄數量:" + groupingCount + ";");
        groupValue.forEach((countKey, countValue) -> System.out.print("英雄攻擊范圍:" + countKey + ";英雄數量:" + countValue + ";"));
        System.out.println();
    });
}

求近戰英雄HP初始值的加總

@Test
fun t3() {
    // 求近戰英雄HP初始值的加總
    val sum = heroes.filter { "近戰" == it.attackRange }
        .map(Hero::getHpStart)
        .filter(Objects::nonNull)
        .reduce(BigDecimal::add)
    println("近戰英雄HP初始值的加總為:$sum")
}

@Test
public void t3() {
    // 求近戰英雄HP初始值的加總
    BigDecimal sum = heroes.stream()
            .filter(hero -> "近戰".equals(hero.getAttackRange()))
            .map(Hero::getHpStart)
            .filter(Objects::nonNull)
            .reduce(BigDecimal.ZERO, BigDecimal::add);
    System.out.println("近戰英雄HP初始值的加總為:" + sum);
}

通過最小串列收集器獲取最小串列

@Test
public void t4() {
    // 通過最小串列收集器獲取最小串列
    List<BigDecimal> minAttackGrowth = heroes.stream()
            .map(Hero::getAttackGrowth)
            .collect(new MinListCollector<>());
    System.out.println(minAttackGrowth);
    List<Hero> minHero = heroes.stream()
            .collect(new MinListCollector<>());
    System.out.println(minHero);
}

import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.function.BinaryOperator;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collector;
import java.util.stream.Collectors;

import static java.util.stream.Collector.Characteristics.*;

/**
 * 最小串列收集器
 *
 * @author switch
 * @since 2020/8/18
 */
public class MinListCollector<T extends Comparable<? super T>> implements Collector<T, List<T>, List<T>> {
    /**
     * 收集器的特性
     *
     * @see Characteristics
     */
    private final static Set<Characteristics> CHARACTERISTICS = Collections.unmodifiableSet(EnumSet.of(IDENTITY_FINISH));
    private final static int ZERO = 0;

    /**
     * 最小值
     */
    private final AtomicReference<T> min = new AtomicReference<>();

    @Override
    public Supplier<List<T>> supplier() {
        // supplier引數用于生成結果容器，容器型別為A
        return ArrayList::new;
    }

    @Override
    public BiConsumer<List<T>, T> accumulator() {
        // accumulator用于消費元素，也就是歸納元素，這里的T就是元素，它會將流中的元素一個一個與結果容器A發生操作
        return (list, element) -> {
            // 獲取最小值
            T minValue = https://www.cnblogs.com/switchvov/p/min.get();
            if (Objects.isNull(minValue)) {
                // 第一次比較
                list.add(element);
                min.set(element);
            } else if (element.compareTo(minValue) < ZERO) {
                // 發現更小的值
                list.clear();
                list.add(element);
                min.compareAndSet(minValue, element);
            } else if (element.compareTo(minValue) == ZERO) {
                // 與最小值相等
                list.add(element);
            }
        };
    }

    @Override
    public BinaryOperator> combiner() {
        // combiner用于兩個兩個合并并行執行的執行緒的執行結果，將其合并為一個最終結果A
        return (left, right) -> {
            // 最小值串列合并
            List leftList = getMinList(left);
            List rightList = getMinList(right);
            leftList.addAll(rightList);
            return leftList;
        };
    }

    private List getMinList(List list) {
        return list.stream()
                .filter(element -> element.compareTo(min.get()) == ZERO)
                .collect(Collectors.toList());
    }

    @Override
    public Function, List> finisher() {
        // finisher用于將之前整合完的結果R轉換成為A
        return Function.identity();
    }

    @Override
    public Set characteristics() {
        // characteristics表示當前Collector的特征值，這是個不可變Set
        return CHARACTERISTICS;
    }
}

優雅的空處理

file

import org.junit.Test;

import java.util.Optional;

/**
 * @author switch
 * @since 2020/8/18
 */
public class OptionalTests {
    @Test
    public void t1() {
        // orElse
        System.out.println(Optional.ofNullable(null).orElse("張三"));
        System.out.println(Optional.ofNullable(null).orElseGet(() -> "李四"));
        System.out.println(Optional.ofNullable("王五").orElseThrow(NullPointerException::new));
    }

    @Test
    public void t2() {
        // isPresent
        Optional<String> name = Optional.ofNullable("張三");
        if (name.isPresent()) {
            System.out.println(name.get());
        }
    }

    @Test
    public void t3() {
        // map
        Optional<Integer> number = Optional.of("123456").map(Integer::valueOf);
        if (number.isPresent()) {
            System.out.println(number.get());
        }
    }

    @Test
    public void t4() {
        // flatMap
        Optional<Integer> number = Optional.of("123456").flatMap(s -> Optional.of(Integer.valueOf(s)));
        if (number.isPresent()) {
            System.out.println(number.get());
        }
    }

    @Test
    public void t5() {
        // 過濾
        String number = "123456";
        String filterNumber = Optional.of(number).filter(s -> !s.equals(number)).orElse("654321");
        System.out.println(filterNumber);
    }
}

新的并發工具類`CompletableFuture`

file

單機批處理多執行緒執行模型

該模型適用于百萬級量級的任務，超過千萬資料，可以考慮分組，多機器并行執行，
基本流程：

從資料庫獲取Id串列

拆分成n個子Id串列

通過子Id串列獲取關聯資料(注意：都需要提供批量查詢介面)

映射到需要處理的Model(提交到CompletableFuture)->處理資料->收集成list)(java 8流式處理)

收集的list進行join操作

收集list

模型

模型原理：Stream+CompletableFuture+lambda

簡要解釋:

CompletableFuture是java8提供的一個工具類，主要是用于異步處理流程編排的，

Stream是java8提供的一個集合流式處理工具類，主要用于資料的流水線處理，

lambda在java中是基于內部匿名類實作的，可以大幅減少重復代碼，

總結：在該模型中Stream用于集合流水線處理、CompletableFuture解決異步編排問題(非阻塞)、lambda簡化代碼，

資料流動

List<List<String>> -> 
Stream<List<String>> -> 
Stream<List<Model>> -> 
Stream<CompletableFuture<List<Model>>> -> 
Stream<CompletableFuture<List<映射型別>>> -> 
List<CompletableFuture<Void>>

案例

ThreadPoolUtil

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.ThreadPoolExecutor;

public final class ThreadPoolUtil {
    public static ThreadPoolTaskExecutor getDefaultExecutor(Integer poolSize, Integer maxPoolSize, Integer queueCapacity) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setAllowCoreThreadTimeOut(true);
        executor.setWaitForTasksToCompleteOnShutdown(true);
        executor.setCorePoolSize(poolSize);
        executor.setMaxPoolSize(maxPoolSize);
        executor.setQueueCapacity(queueCapacity);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }
}

ThreadPoolConfig

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
public class ThreadPoolConfig {
    /**
     * 計算規則：N(thread) = N(cpu) * U(cpu) * (1 + w/c)
     * N(thread)：執行緒池大小
     * N(cpu)：處理器核數
     * U(cpu)：期望CPU利用率（該值應該介于0和1之間）
     * w/c：是等待時間與計算時間的比率，比如說IO操作即為等待時間，計算處理即為計算時間
     */
    private static final Integer TASK_POOL_SIZE = 50;
    private static final Integer TASK_MAX_POOL_SIZE = 100;
    private static final Integer TASK_QUEUE_CAPACITY = 1000;

    @Bean("taskExecutor")
    public ThreadPoolTaskExecutor taskExecutor() {
        return ThreadPoolUtil.getDefaultExecutor(TASK_POOL_SIZE, TASK_MAX_POOL_SIZE, TASK_QUEUE_CAPACITY);
    }
}

#getFuturesStream

public Stream<CompletableFuture<List<Model>>> getFuturesStream(List<List<String>> idSubLists) {
    return idSubLists.stream()
        .map(ids -> 
            CompletableFuture.supplyAsync(() -> modelService.listByIds(ids), taskExecutor)
        );
}

#standardisation

public void standardisation() {
    List<CompletableFuture<Void>> batchFutures = getFuturesStream(idSubLists)
            .map(future -> future.thenApply(this::listByNormalize))
            .map(future -> future.thenAccept(modelService::batchUpdateData))
            .collect(Collectors.toList());
    List<Void> results = batchFutures.stream()
            .map(CompletableFuture::join)
            .collect(Collectors.toList());
}

調整執行緒池的大小

《Java并發編程實戰》一書中，Brian Goetz和合著者們為執行緒池大小的優化提供了不少中肯的建議，這非常重要，如果執行緒池中執行緒的數量過多，最終它們會競爭稀缺的處理器和記憶體資源，浪費大量的時間在背景關系切換上，反之，如果執行緒的數目過少，正如你的應用所面臨的情況，處理器的一些核可能就無法充分利用，Brian Goetz建議，執行緒池大小與處理器的利用率之比可以使用下面的公式進行估算：
$$N_{threads} = N_{CPU} * U_{CPU} * (1 + \frac{W}{C})$$

其中：

$N_{CPU}$是處理器的核的數目，可以通過Runtime.getRuntime().availableProcessors()得到

$U_{CPU}$是期望的CPU利用率（該值應該介于0和1之間）

$\frac{W}{C}$是等待時間與計算時間的比率，比如說IO操作即為等待時間，計算處理即為計算時間

并行——使用流還是CompletableFutures？

對集合進行并行計算有兩種方式：要么將其轉化為并行流，利用map這樣的操作開展作業，要么列舉出集合中的每一個元素，創建新的執行緒，在CompletableFuture內對其進行操作，后者提供了更多的靈活性，可以調整執行緒池的大小，而這能幫助確保整體的計算不會因為執行緒都在等待I/O而發生阻塞，

使用這些API的建議如下：

如果進行的是計算密集型的操作，并且沒有I/O，那么推薦使用Stream介面，因為實作簡單，同時效率也可能是最高的（如果所有的執行緒都是計算密集型的，那就沒有必要創建比處理器核數更多的執行緒），

反之，如果并行的作業單元還涉及等待I/O的操作（包括網路連接等待），那么使用CompletableFuture靈活性更好，可以依據等待/計算，或者$\frac{W}{C}$的比率設定需要使用的執行緒數，這種情況不使用并行流的另一個原因是，處理流的流水線中如果發生I/O等待，流的延遲特性很難判斷到底什么時候觸發了等待，

日期和時間API

file

使用指南：https://www.yuque.com/docs/share/ee5ef8a7-d261-4593-bd08-2a7a7d2c11ca?#（密碼：gtag）《時區工具類使用指南》

專案地址

GitHub：java8-fluent

參考

Java 8 實戰學習筆記
Java 8 函式式編程學習筆記
深入理解Java函式式編程和Streams API

分享并記錄所學所見

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/237861.html

標籤：Java

上一篇：MySQL（二）：快速理解MySQL資料庫索引

下一篇：基于 Spring Boot 的車牌識別系統（附專案地址）

怎樣用Java 8優雅的開發業務

怎樣用Java 8優雅的開發業務

函式式編程

流式編程

基本原理

案例

優雅的空處理

新的并發工具類CompletableFuture

單機批處理多執行緒執行模型

模型

案例

調整執行緒池的大小

并行——使用流還是CompletableFutures？

日期和時間API

專案地址

參考

新的并發工具類`CompletableFuture`