复制代码// 无参数，无返回值
() -> log.info("Lambda")

 // 有参数，有返回值
(int a, int b) -> { a+b }

复制代码log.info("Lambda");

private int plus(int a, int b){
   return a+b;
}

复制代码new Thread(new Runnable() {
    @Override
    public void run() {
        System.out.println("快速新建并启动一个线程");
    }
}).start();

复制代码new Thread(()->{
    System.out.println("快速新建并启动一个线程");
}).start();

复制代码Function<String, Integer> s = Integer::parseInt;
Integer i = s.apply("10");

复制代码Comparator<Integer> comparator = Integer::compare;
int result = comparator.compare(100,10);

复制代码IntBinaryOperator intBinaryOperator = Integer::compare;
int result = intBinaryOperator.applyAsInt(10,100);

复制代码@FunctionalInterface
public interface Function<T, R> {
    R apply(T t);
}

复制代码public static int parseInt(String s) throws NumberFormatException {
    return parseInt(s,10);
}

复制代码public static int compare(int x, int y) {
    return (x < y) ? -1 : ((x == y) ? 0 : 1);
}

复制代码@FunctionalInterface
public interface Comparator<T> {
    int compare(T o1, T o2);
}

@FunctionalInterface
public interface IntBinaryOperator {
    int applyAsInt(int left, int right);
}

复制代码@FunctionalInterface
public interface KiteFunction<T, R, S> {

    /**
     * 定义一个双参数的方法
     * @param t
     * @param s
     * @return
     */
    R run(T t,S s);
}

复制代码public class FunctionTest {
    public static String DateFormat(LocalDateTime dateTime, String partten) {
        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(partten);
        return dateTime.format(dateTimeFormatter);
    }
}

复制代码KiteFunction<LocalDateTime,String,String> functionDateFormat = FunctionTest::DateFormat;
String dateString = functionDateFormat.run(LocalDateTime.now(),"yyyy-MM-dd HH:mm:ss");

复制代码public static void main(String[] args) throws Exception {
  
    String dateString = new KiteFunction<LocalDateTime, String, String>() {
        @Override
        public String run(LocalDateTime localDateTime, String s) {
            DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(s);
            return localDateTime.format(dateTimeFormatter);
        }
    }.run(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss");
    System.out.println(dateString);
}

复制代码public static void main(String[] args) throws Exception {

        KiteFunction<LocalDateTime, String, String> functionDateFormat = (LocalDateTime dateTime, String partten) -> {
            DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern(partten);
            return dateTime.format(dateTimeFormatter);
        };
        String dateString = functionDateFormat.run(LocalDateTime.now(), "yyyy-MM-dd HH:mm:ss");
        System.out.println(dateString);
}

复制代码public interface Stream<T> extends BaseStream<T, Stream<T>> {
  
  Stream<T> filter(Predicate<? super T> predicate);
  
  // 其他接口
}

复制代码private static void createStream(){
    Stream<String> stringStream = Stream.of("a","b","c");
}

复制代码private static void concatStream(){
    Stream<String> a = Stream.of("a","b","c");
    Stream<String> b = Stream.of("d","e");
    Stream<String> c = Stream.concat(a,b);
}

复制代码private static void max(){
    Stream<Integer> integerStream = Stream.of(2, 2, 100, 5);
    Integer max = integerStream.max(Integer::compareTo).get();
    System.out.println(max);
}

复制代码private static void max(){
    Stream<Integer> integerStream = Stream.of(2, 2, 100, 5);
    Comparator<Integer> comparator =  (x, y) -> (x.intValue() < y.intValue()) ? -1 : ((x.equals(y)) ? 0 : 1);
    Integer max = integerStream.max(comparator).get();
    System.out.println(max);
}

复制代码Stream<String> a = Stream.of("a", "b", "c");
long x = a.count();

复制代码private static void peek() {
    Stream<String> a = Stream.of("a", "b", "c");
    List<String> list = a.peek(e->System.out.println(e.toUpperCase())).collect(Collectors.toList());
}

复制代码private static void forEach() {
    Stream<String> a = Stream.of("a", "b", "c");
    a.forEach(e->System.out.println(e.toUpperCase()));
}

复制代码Stream<String> a = Stream.of("a", "b", "c");
a.parallel().forEach(e->System.out.println(e.toUpperCase()));

复制代码Stream<String> a = Stream.of("a", "b", "c");
a.parallel().forEachOrdered(e->System.out.println(e.toUpperCase()));

复制代码private static void limit() {
    Stream<String> a = Stream.of("a", "b", "c");
    a.limit(2).forEach(e->System.out.println(e));
}

复制代码private static void skip() {
    Stream<String> a = Stream.of("a", "b", "c");
    a.skip(2).forEach(e->System.out.println(e));
}

复制代码private static void distinct() {
    Stream<String> a = Stream.of("a", "b", "c","b");
    a.distinct().forEach(e->System.out.println(e));
}

复制代码private static void sorted() {
    Stream<String> a = Stream.of("a", "c", "b");
    a.sorted().forEach(e->System.out.println(e));
}

复制代码private static void sortedWithComparator() {
    Stream<String> a = Stream.of("a1", "c6", "b3");
    a.sorted((x,y)->Integer.parseInt(x.substring(1))>Integer.parseInt(y.substring(1))?1:-1).forEach(e->System.out.println(e));
}

复制代码private static List<User> getUserData() {
    Random random = new Random();
    List<User> users = new ArrayList<>();
    for (int i = 1; i <= 10; i++) {
        User user = new User();
        user.setUserId(i);
        user.setUserName(String.format("古时的风筝 %s 号", i));
        user.setAge(random.nextInt(100));
        user.setGender(i % 2);
        user.setPhone("18812021111");
        user.setAddress("无");
        users.add(user);
    }
    return users;
}

复制代码private static void filter(){
    List<User> users = getUserData();
    Stream<User> stream = users.stream();
    stream.filter(user -> user.getGender().equals(0) && user.getAge()>50).forEach(e->System.out.println(e));

    /**
     *等同于下面这种形式 匿名内部类
     */
//    stream.filter(new Predicate<User>() {
//        @Override
//        public boolean test(User user) {
//            return user.getGender().equals(0) && user.getAge()>50;
//        }
//    }).forEach(e->System.out.println(e));
}

复制代码<R> Stream<R> map(Function<? super T, ? extends R> mapper);

复制代码@FunctionalInterface
public interface Function<T, R> {

    /**
     * Applies this function to the given argument.
     *
     * @param t the function argument
     * @return the function result
     */
    R apply(T t);
}

复制代码private static void map(){
    List<User> users = getUserData();
    Stream<User> stream = users.stream();
    List<UserDto> userDtos = stream.map(user -> dao2Dto(user)).collect(Collectors.toList());
}

private static UserDto dao2Dto(User user){
    UserDto dto = new UserDto();
    BeanUtils.copyProperties(user, dto);
    //其他额外处理
    return dto;
}

复制代码private static void flatMap(){
    List<User> users = getUserData();
    List<User> users1 = getUserData();
    List<List<User>> userList = new ArrayList<>();
    userList.add(users);
    userList.add(users1);
    Stream<List<User>> stream = userList.stream();
    List<UserDto> userDtos = stream.flatMap(subUserList->subUserList.stream()).map(user -> dao2Dto(user)).collect(Collectors.toList());
}

复制代码<R, A> R collect(Collector<? super T, A, R> collector);

复制代码private static void collect(){
    Stream<Integer> integerStream = Stream.of(1,2,5,7,8,12,33);
    List<Integer> list = integerStream.filter(s -> s.intValue()>7).collect(Collectors.toList());
}

复制代码private static void collect(){
    Stream<Integer> integerStream = Stream.of(1,2,5,7,8,12,33);
    List<Integer> list = integerStream.filter(s -> s.intValue()>7).collect(ArrayList::new, ArrayList::add,
            ArrayList::addAll);
}

复制代码// 返回 userId:List<User>
Map<String,List<User>> map = user.stream().collect(Collectors.groupingBy(User::getUserId));

// 返回 userId:每组个数
Map<String,Long> map = user.stream().collect(Collectors.groupingBy(User::getUserId,Collectors.counting()));

复制代码@FunctionalInterface
public interface IntFunction<R> {

    /**
     * Applies this function to the given argument.
     *
     * @param value the function argument
     * @return the function result
     */
    R apply(int value);
}

复制代码private static void toArray() {
    List<User> users = getUserData();
    Stream<User> stream = users.stream();
    User[] userArray = stream.filter(user -> user.getGender().equals(0) && user.getAge() > 50).toArray(User[]::new);
}

复制代码private static void reduce(){
    Stream<Integer> integerStream = Stream.of(1,2,5,7,8,12,33);
    Integer sum = integerStream.reduce(0,(x,y)->x+y);
    System.out.println(sum);
}

复制代码if (
  "mediaDevices" in navigator &&
  "getUserMedia" in navigator.mediaDevices
) {
  // ok, 浏览器支持它
}

复制代码const videoStream = await navigator.mediaDevices.getUserMedia({ video: true });

复制代码const constraints = {
  video: {
    width: {
      min: 1280,
      ideal: 1920,
      max: 2560,
    },
    height: {
      min: 720,
      ideal: 1080,
      max: 1440,
    },
  },
};

const videoStream = await navigator.mediaDevices.getUserMedia(constraints);

复制代码// 页面中有一个 <video autoplay id="video"></video> 标签
const video = document.querySelector("#video");
const videoStream = await navigator.mediaDevices.getUserMedia(constraints);
video.srcObject = videoStream;

复制代码const constraints = {
  video: {
    width: { ... },
    height: { ... },
    facingMode: "environment"
  },
};

复制代码videoStream.getTracks().forEach((track) => {
  track.stop();
});

复制代码// 页面中有一个 <canvas id="canvas"></canvas> 标签
const canvas = document.querySelector("#canvas");
canvas.width = video.videoWidth;
canvas.height = video.videoHeight;
canvas.getContext("2d").drawImage(video, 0, 0);

复制代码const img = document.createElement("img");
img.src = canvas.toDataURL("image/png");
screenshotsContainer.prepend(img);

复制代码@ChannelHandler.Sharable // 标识一个 channelHandler 可以被多个 channel 安全地调用。
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    // 当有入站消息时该方法就会调用
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buffer = (ByteBuf)msg;
        System.out.println("服务器收到消息:" + buffer.toString(CharsetUtil.UTF_8));
        // 将接收到的消息写给发送者，而不冲刷出站消息。
        ctx.write(buffer);
    }

    @Override
    // channelRead消费完读取的数据的时候被触发
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        // 将未决消息冲刷到远程节点，并且关闭该 channel
        ChannelFuture channelFuture = ctx.writeAndFlush(Unpooled.EMPTY_BUFFER);
        channelFuture.addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    // 在读操作时处理异常
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 打印异常栈，并关闭该 channel
        cause.printStackTrace();
        ctx.close();
    }
}

复制代码public class NettyServer {
    public static void main(String[] args) {
        NettyServer nettyServer = new NettyServer();
        nettyServer.start(8888);
    }

    public void start(int port){
        // 处理TCP连接请求
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 处理I/O事件
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            // 用于引导和绑定服务器
            ServerBootstrap bootstrap = new ServerBootstrap();
            //将上面的线程组加入到 bootstrap 中
            bootstrap.group(bossGroup,workGroup)
                    //将通道设置为异步的通道
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            // 因为 NettyServerHandler 被标注为 @Sharable，所以可以使用相同的实例
                            socketChannel.pipeline().addLast(new NettyServerHandler());
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG,200)
                    .childOption(ChannelOption.SO_KEEPALIVE,true);
            // 异步绑定服务器，调用 sync() 方法阻塞等待直到绑定完成。
            ChannelFuture future = bootstrap.bind(port).sync();
            // 获取 channel 的 closeFuture，并且阻塞直到它完成。
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

复制代码@ChannelHandler.Sharable
public class NettyClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
       ctx.writeAndFlush(Unpooled.copiedBuffer("netty 活跃", CharsetUtil.UTF_8));
    }

    // 记录已接收的消息存储
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        System.out.println("客户端接收消息：" + msg.toString(CharsetUtil.UTF_8));
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

复制代码public class NettyClient {
    public static void main(String[] args) {
        NettyClient nettyClient = new NettyClient();
        nettyClient.connect("localhost", 8888);
    }

    public void connect(String hostname,int port) {
        // 处理TCP连接请求
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            // 用于引导和绑定服务器
            Bootstrap bootstrap = new Bootstrap();
            //将上面的线程组加入到 bootstrap 中
            bootstrap.group(group)
                	//将通道设置为异步的通道
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline()
                                    .addLast(new NettyClientHandler());
                        }
                    })
                    .option(ChannelOption.TCP_NODELAY,true);

            // 连接到远程节点，阻塞等待直到连接完成。
            ChannelFuture future = bootstrap.connect(hostname, port).sync();
            // 阻塞，直到 channel 关闭。
            future.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 关闭线程池并释放所有资源。
            group.shutdownGracefully();
        }
    }
}

复制代码// NioServerSocketChannel构造方法
public NioServerSocketChannel(ServerSocketChannel channel) {
    super(null, channel, SelectionKey.OP_ACCEPT);
    config = new NioServerSocketChannelConfig(this, javaChannel().socket());
}

复制代码NioServerSocketChannel channel = new NioServerSocketChannel();
ServerSocketChannelConfig config = channel.config();
config.setOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
// 针对每个ChannelOption选项，netty还提供了对应的方法，比如上面的设置可以这样代替
// config.setAllocator(PooledByteBufAllocator.DEFAULT);
// 设置通道
bootstrap.channel(channel.getClass());

复制代码@Test
public void testFramesDecoded(){
    ByteBuf buf= Unpooled.buffer();
    for (int i=0;i<9;i++){
        buf.writeByte(i);
    }
    ByteBuf input=buf.duplicate();
    EmbeddedChannel channel=new EmbeddedChannel(
        new FixedLengthFrameDecoder(3)
    );
    assertTrue(channel.writeInbound(input.retain()));
    assertTrue(channel.finish());    
    //读取消息
    ByteBuf read=channel.readInbound();
    assertEquals(buf.readSlice(3),read);
    read.release();

    read = (ByteBuf) channel.readInbound();
    assertEquals(buf.readSlice(3), read);
    read.release();
    read = (ByteBuf) channel.readInbound();
    assertEquals(buf.readSlice(3), read);
    read.release();
    assertNull(channel.readInbound());
    buf.release();
}

复制代码// ByteToCharDecoder是自定义的编码器，CharToByteEncoder是自定义的解码器
public class CombinedByteCharCodec extends
    CombinedChannelDuplexHandler<ByteToCharDecoder, CharToByteEncoder> {
    public CombinedByteCharCodec() {
    super(new ByteToCharDecoder(), new CharToByteEncoder());
    }
}

复制代码public static void heapBuffer() {
    // 创建Java堆缓冲区
    ByteBuf heapBuf = Unpooled.buffer(); 
    if (heapBuf.hasArray()) { // 判断是否有支撑数组
        byte[] array = heapBuf.array(); // 获取该支撑数组引用
        int offset = heapBuf.arrayOffset() + heapBuf.readerIndex(); // 计算第一个字节偏移量
        int length = heapBuf.readableBytes(); // 获取可读字节数
        handleArray(array, offset, length); // 调用自己的方法
    }
}

复制代码public static void directBuffer() {
    ByteBuf directBuf = Unpooled.directBuffer();
    if (!directBuf.hasArray()) { // 如果不是堆缓冲区
        int length = directBuf.readableBytes(); // 获取可读字节数
        byte[] array = new byte[length]; // 分配一个新数组来保存数据
        directBuf.getBytes(directBuf.readerIndex(), array); // 将数据复制到新数组
        handleArray(array, 0, length); // 调用自己的方法
    }
}

复制代码public static void byteBufComposite() {
    // 复合缓冲区，只是提供一个视图
    CompositeByteBuf messageBuf = Unpooled.compositeBuffer();
    ByteBuf headerBuf = Unpooled.buffer();
    ByteBuf bodyBuf = Unpooled.directBuffer();
    messageBuf.addComponents(headerBuf, bodyBuf); // 将ByteBuf实例追加到CompositeByteBuf
    messageBuf.removeComponent(0); // 移除索引位置为0的缓冲区
    for (ByteBuf buf : messageBuf) { // 遍历缓冲区
        System.out.println(buf.toString());
    }
}

复制代码public static void byteBufRelativeAccess() {
    ByteBuf buffer = Unpooled.buffer();
    for (int i = 0; i < buffer.capacity(); i++) {
        byte b = buffer.getByte(i);// 不改变readerIndex值
        System.out.println((char) b);
    }
}

复制代码public static void main(String[] args) {
        // 创建线程池。这里只是临时测试，开发规范...
        ExecutorService executor = Executors.newFixedThreadPool(10);

        // 提交任务到线程池中执行。
        executor.submit(new Runnable() {

            @Override
            public void run() {
                System.out.println("听说我被异步调用了");
            }

        });
    }

复制代码<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>lab-29-async-demo</artifactId>

    <dependencies>
        <!-- 引入 Spring Boot 依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!-- 方便等会写单元测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

复制代码@SpringBootApplication
@EnableAsync // 开启 @Async 的支持
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

复制代码package cn.iocoder.springboot.lab29.asynctask.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.AsyncResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

@Service
public class DemoService {

    private Logger logger = LoggerFactory.getLogger(getClass());
    
    public Integer execute01() {
        logger.info("[execute01]");
        sleep(10);
        return 1;
    }

    public Integer execute02() {
        logger.info("[execute02]");
        sleep(5);
        return 2;
    }

    private static void sleep(int seconds) {
        try {
            Thread.sleep(seconds * 1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Async
    public Integer zhaoDaoNvPengYou(Integer a, Integer b) {
        throw new RuntimeException("程序员不需要女朋友");
    }

}

复制代码@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class DemoServiceTest {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private DemoService demoService;

    @Test
    public void task01() {
        long now = System.currentTimeMillis();
        logger.info("[task01][开始执行]");

        demoService.execute01();
        demoService.execute02();

        logger.info("[task01][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }
}

复制代码2020-06-02 09:16:03.391  INFO 3108 --- [      main] c.i.s.l.a.service.DemoServiceTest        : [task01][开始执行]
2020-06-02 09:16:03.402  INFO 3108 --- [      main] c.i.s.l.asynctask.service.DemoService    : [execute01]
2020-06-02 09:16:13.403  INFO 3108 --- [      main] c.i.s.l.asynctask.service.DemoService    : [execute02]
2020-06-02 09:16:18.403  INFO 3108 --- [      main] c.i.s.l.a.service.DemoServiceTest        : [task01][结束执行，消耗时长 15012 毫秒]

复制代码	@Async
    public Integer execute01Async() {
        return this.execute01();
    }

    @Async
    public Integer execute02Async() {
        return this.execute02();
    }

复制代码	@Test
    public void task02() {
        long now = System.currentTimeMillis();
        logger.info("[task02][开始执行]");

        demoService.execute01Async();
        demoService.execute02Async();

        logger.info("[task02][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }

复制代码2020-06-02 10:57:41.643  INFO 14416 --- [main] c.i.s.l.a.service.DemoServiceTest        : [task02][开始执行]
2020-06-02 10:57:41.675  INFO 14416 --- [main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-06-02 10:57:41.682  INFO 14416 --- [main] c.i.s.l.a.service.DemoServiceTest        : [task02][结束执行，消耗时长 39 毫秒]

复制代码	@Async
    public Future<Integer> execute01AsyncWithFuture() {
        return AsyncResult.forValue(this.execute01());
    }

    @Async
    public Future<Integer> execute02AsyncWithFuture() {
        return AsyncResult.forValue(this.execute02());
    }

复制代码	@Test
    public void task03() throws ExecutionException, InterruptedException {
        long now = System.currentTimeMillis();
        logger.info("[task03][开始执行]");

        // 执行任务
        Future<Integer> execute01Result = demoService.execute01AsyncWithFuture();
        Future<Integer> execute02Result = demoService.execute02AsyncWithFuture();
        // 阻塞等待结果
        execute01Result.get();
        execute02Result.get();

        logger.info("[task03][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }

复制代码2020-06-02 13:56:43.955  INFO 7828 --- [ main] c.i.s.l.a.service.DemoServiceTest        : [task03][开始执行]
2020-06-02 13:56:43.987  INFO 7828 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor  : Initializing ExecutorService 'applicationTaskExecutor'
2020-06-02 13:56:44.008  INFO 7828 --- [ task-1] c.i.s.l.asynctask.service.DemoService    : [execute01]
2020-06-02 13:56:44.008  INFO 7828 --- [ task-2] c.i.s.l.asynctask.service.DemoService    : [execute02]
2020-06-02 13:56:54.008  INFO 7828 --- [ main] c.i.s.l.a.service.DemoServiceTest        : [task03][结束执行，消耗时长 10053 毫秒]

复制代码spring:
  task:
    # Spring 执行器配置，对应 TaskExecutionProperties 配置类。对于 Spring 异步任务，会使用该执行器。
    execution:
      thread-name-prefix: task- # 线程池的线程名的前缀。默认为 task- ，建议根据自己应用来设置
      pool: # 线程池相关
        core-size: 8 # 核心线程数，线程池创建时候初始化的线程数。默认为 8 。
        max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE
        keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒
        queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE 。
        allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true 。
      shutdown:
        await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true
        await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0 ，根据自己应用来设置

复制代码	/**
	 * Create a new async result which exposes the given value from {@link Future#get()}.
	 * @param value the value to expose
	 * @since 4.2
	 * @see Future#get()
	 */
	public static <V> ListenableFuture<V> forValue(V value) {
		return new AsyncResult<>(value, null);
	}

复制代码	/**
	 * Create a new async result which exposes the given exception as an
	 * {@link ExecutionException} from {@link Future#get()}.
	 * @param ex the exception to expose (either an pre-built {@link ExecutionException}
	 * or a cause to be wrapped in an {@link ExecutionException})
	 * @since 4.2
	 * @see ExecutionException
	 */
	public static <V> ListenableFuture<V> forExecutionException(Throwable ex) {
		return new AsyncResult<>(null, ex);
	}

复制代码public class AsyncResult<V> implements ListenableFuture<V>

复制代码public interface ListenableFuture<T> extends Future<T> {

    // 添加回调方法，统一处理成功和异常的情况。
	void addCallback(ListenableFutureCallback<? super T> callback);

	// 添加成功和失败的回调方法，分别处理成功和异常的情况。
	void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback);


	// 将 ListenableFuture 转换成 JDK8 提供的 CompletableFuture 。
    // 这样，后续我们可以使用 ListenableFuture 来设置回调
	default CompletableFuture<T> completable() {
		CompletableFuture<T> completable = new DelegatingCompletableFuture<>(this);
		addCallback(completable::complete, completable::completeExceptionally);
		return completable;
	}

}

复制代码public interface Future<V> {

    // 如果任务还没开始，执行 cancel(...) 方法将返回 false；
    // 如果任务已经启动，执行 cancel(true) 方法将以中断执行此任务线程的方式来试图停止任务，如果停止成功，返回 true ；
    // 当任务已经启动，执行 cancel(false) 方法将不会对正在执行的任务线程产生影响(让线程正常执行到完成)，此时返回 false ；
    // 当任务已经完成，执行 cancel(...) 方法将返回 false 。
    // mayInterruptRunning 参数表示是否中断执行中的线程。
    boolean cancel(boolean mayInterruptIfRunning);

    // 如果任务完成前被取消，则返回 true 。
    boolean isCancelled();

    // 如果任务执行结束，无论是正常结束或是中途取消还是发生异常，都返回 true 。
    boolean isDone();

    // 获取异步执行的结果，如果没有结果可用，此方法会阻塞直到异步计算完成。
    V get() throws InterruptedException, ExecutionException;
    
	// 获取异步执行结果，如果没有结果可用，此方法会阻塞，但是会有时间限制，如果阻塞时间超过设定的 timeout 时间，该方法将抛出异常。
    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

复制代码	@Override
	public void addCallback(ListenableFutureCallback<? super V> callback) {
		addCallback(callback, callback);
	}

	@Override
	public void addCallback(SuccessCallback<? super V> successCallback, FailureCallback failureCallback) {
		try {
			if (this.executionException != null) { // 《1》
				failureCallback.onFailure(exposedException(this.executionException));
			}
			else { // 《2》
				successCallback.onSuccess(this.value);
			}
		}
		catch (Throwable ex) { // 《3》
			// Ignore
		}
	}

// 从 ExecutionException 中，获得原始异常。
private static Throwable exposedException(Throwable original) {
    if (original instanceof ExecutionException) {
        Throwable cause = original.getCause();
        if (cause != null) {
            return cause;
        }
    }
    return original;
}

复制代码public interface ListenableFutureCallback<T> extends SuccessCallback<T>, FailureCallback

复制代码// AsyncResult.java

@Override
public boolean cancel(boolean mayInterruptIfRunning) {
    return false; // 因为是 AsyncResult 是执行结果，所以直接返回 false 表示取消失败。
}

@Override
public boolean isCancelled() {
    return false; // 因为是 AsyncResult 是执行结果，所以直接返回 false 表示未取消。
}

@Override
public boolean isDone() {
    return true; // 因为是 AsyncResult 是执行结果，所以直接返回 true 表示已完成。
}

@Override
@Nullable
public V get() throws ExecutionException {
    // 如果发生异常，则抛出该异常。
    if (this.executionException != null) {
        throw (this.executionException instanceof ExecutionException ?
                (ExecutionException) this.executionException :
                new ExecutionException(this.executionException));
    }
    // 如果执行成功，则返回该 value 结果
    return this.value;
}

@Override
@Nullable
public V get(long timeout, TimeUnit unit) throws ExecutionException {
    return get();
}

复制代码private final ListenableFutureCallbackRegistry<T> callbacks = new ListenableFutureCallbackRegistry<T>();

@Override
public void addCallback(ListenableFutureCallback<? super T> callback) {
    this.callbacks.addCallback(callback);
}

@Override
public void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback) {
    this.callbacks.addSuccessCallback(successCallback);
    this.callbacks.addFailureCallback(failureCallback);
}

复制代码@Override
	protected void done() {
		Throwable cause;
		try {
            // 获得执行结果
			T result = get();
            // 执行成功，执行成功的回调
			this.callbacks.success(result);
			return;
		}
		catch (InterruptedException ex) { // 如果有中断异常 InterruptedException异常，则打断当前线程，直接返回
			Thread.currentThread().interrupt();
			return;
		}
		catch (ExecutionException ex) { // 如果有 ExecutionException 异常，获取真实异常，并设置到cause中
			cause = ex.getCause();
			if (cause == null) {
				cause = ex;
			}
		}
		catch (Throwable ex) {
			cause = ex; // 设置异常到 cause 中
		}
         // 执行异常，执行异常的回调
		this.callbacks.failure(cause);
	}

复制代码@Async
    public ListenableFuture<Integer> execute01AsyncWithListenableFuture() {
        try {
            //int i = 1 / 0;
            return AsyncResult.forValue(this.execute02());
        } catch (Throwable ex) {
            return AsyncResult.forExecutionException(ex);
        }
    }

复制代码@Test
    public void task04() throws ExecutionException, InterruptedException {
        long now = System.currentTimeMillis();
        logger.info("[task04][开始执行]");

        // <1> 执行任务
        ListenableFuture<Integer> execute01Result = demoService.execute01AsyncWithListenableFuture();
        logger.info("[task04][execute01Result 的类型是：({})]",execute01Result.getClass().getSimpleName());
        execute01Result.addCallback(new SuccessCallback<Integer>() { // <2.1> 增加成功的回调

            @Override
            public void onSuccess(Integer result) {
                logger.info("[onSuccess][result: {}]", result);
            }

        }, new FailureCallback() { // <2.1> 增加失败的回调

            @Override
            public void onFailure(Throwable ex) {
                logger.info("[onFailure][发生异常]", ex);
            }

        });
        execute01Result.addCallback(new ListenableFutureCallback<Integer>() { // <2.2> 增加成功和失败的统一回调

            @Override
            public void onSuccess(Integer result) {
                logger.info("[onSuccess][result: {}]", result);
            }

            @Override
            public void onFailure(Throwable ex) {
                logger.info("[onFailure][发生异常]", ex);
            }

        });
        // <3> 阻塞等待结果
        execute01Result.get();

        logger.info("[task04][结束执行，消耗时长 {} 毫秒]", System.currentTimeMillis() - now);
    }

复制代码2020-06-08 14:13:16.738  INFO 5060 --- [  main] c.i.s.l.a.service.DemoServiceTest : [task04][execute01Result 的类型是：(ListenableFutureTask)]

复制代码2020-06-08 14:13:21.752  INFO 5060 --- [   main] c.i.s.l.a.service.DemoServiceTest   : [task04][结束执行，消耗时长 5057 毫秒]
2020-06-08 14:13:21.752  INFO 5060 --- [ task-1] c.i.s.l.a.service.DemoServiceTest   : [onSuccess][result: 2]
2020-06-08 14:13:21.752  INFO 5060 --- [ task-1] c.i.s.l.a.service.DemoServiceTest   : [onSuccess][result: 2]

复制代码@Component
public class GlobalAsyncExceptionHandler implements AsyncUncaughtExceptionHandler {

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void handleUncaughtException(Throwable ex, Method method, Object... params) {
        logger.error("[handleUncaughtException][method({}) params({}) 发生异常]",
                method, params, ex);
    }

}

复制代码// AsyncExecutionAspectSupport.java

protected void handleError(Throwable ex, Method method, Object... params) throws Exception {
    // 重点！！！如果返回类型是 Future ，则直接抛出该异常。
    if (Future.class.isAssignableFrom(method.getReturnType())) {
        ReflectionUtils.rethrowException(ex);
    } else {
        // 否则，交给 AsyncUncaughtExceptionHandler 来处理。
        // Could not transmit the exception to the caller with default executor
        try {
            this.exceptionHandler.obtain().handleUncaughtException(ex, method, params);
        } catch (Throwable ex2) {
            logger.warn("Exception handler for async method '" + method.toGenericString() +
                    "' threw unexpected exception itself", ex2);
        }
    }
}

复制代码@Configuration
@EnableAsync // 开启 @Async 的支持
public class AsyncConfig implements AsyncConfigurer {

    @Autowired
    private GlobalAsyncExceptionHandler exceptionHandler;

    @Override
    public Executor getAsyncExecutor() {
        return null;
    }

    @Override
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
        return exceptionHandler;
    }

}

复制代码@Async
public Integer zhaoDaoNvPengYou(Integer a, Integer b) {
    throw new RuntimeException("异步全局异常");
}

复制代码 @Test
    public void testZhaoDaoNvPengYou() throws InterruptedException {
        demoService.zhaoDaoNvPengYou(1, 2);

        // sleep 1 秒，保证异步调用的执行
        Thread.sleep(1000);
    }

复制代码2020-06-08 15:26:35.120 ERROR 11388 --- [         task-1] .i.s.l.a.c.a.GlobalAsyncExceptionHandler : [handleUncaughtException][method(public java.lang.Integer cn.iocoder.springboot.lab29.asynctask.service.DemoService.zhaoDaoNvPengYou(java.lang.Integer,java.lang.Integer)) params([1, 2]) 发生异常]

java.lang.RuntimeException: 异步全局异常

复制代码<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.2.1.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>lab-29-async-demo</artifactId>

    <dependencies>
        <!-- 引入 Spring Boot 依赖 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <!-- 方便等会写单元测试 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>

</project>

复制代码spring:
  task:
    # Spring 执行器配置，对应 TaskExecutionProperties 配置类。对于 Spring 异步任务，会使用该执行器。
    execution-one:
      thread-name-prefix: task-one- # 线程池的线程名的前缀。默认为 task- ，建议根据自己应用来设置
      pool: # 线程池相关
        core-size: 8 # 核心线程数，线程池创建时候初始化的线程数。默认为 8 。
        max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE
        keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒
        queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE 。
        allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true 。
      shutdown:
        await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true
        await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0 ，根据自己应用来设置
    # Spring 执行器配置，对应 TaskExecutionProperties 配置类。对于 Spring 异步任务，会使用该执行器。
    execution-two:
      thread-name-prefix: task-two- # 线程池的线程名的前缀。默认为 task- ，建议根据自己应用来设置
      pool: # 线程池相关
        core-size: 8 # 核心线程数，线程池创建时候初始化的线程数。默认为 8 。
        max-size: 20 # 最大线程数，线程池最大的线程数，只有在缓冲队列满了之后，才会申请超过核心线程数的线程。默认为 Integer.MAX_VALUE
        keep-alive: 60s # 允许线程的空闲时间，当超过了核心线程之外的线程，在空闲时间到达之后会被销毁。默认为 60 秒
        queue-capacity: 200 # 缓冲队列大小，用来缓冲执行任务的队列的大小。默认为 Integer.MAX_VALUE 。
        allow-core-thread-timeout: true # 是否允许核心线程超时，即开启线程池的动态增长和缩小。默认为 true 。
      shutdown:
        await-termination: true # 应用关闭时，是否等待定时任务执行完成。默认为 false ，建议设置为 true
        await-termination-period: 60 # 等待任务完成的最大时长，单位为秒。默认为 0 ，根据自己应用来设置

复制代码@Configuration
@EnableAsync // 开启 @Async 的支持
public class AsyncConfig
{

    public static final String EXECUTOR_ONE_BEAN_NAME = "executor-one";
    public static final String EXECUTOR_TWO_BEAN_NAME = "executor-two";

    @Configuration
    public static class ExecutorOneConfiguration
    {

        @Bean(name = EXECUTOR_ONE_BEAN_NAME + "-properties")
        @Primary
        @ConfigurationProperties(prefix = "spring.task.execution-one")
        // 读取 spring.task.execution-one 配置到 TaskExecutionProperties 对象
        public TaskExecutionProperties taskExecutionProperties()
        {
            return new TaskExecutionProperties();
        }

        @Bean(name = EXECUTOR_ONE_BEAN_NAME)
        public ThreadPoolTaskExecutor threadPoolTaskExecutor()
        {
            // 创建 TaskExecutorBuilder 对象
            TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());
            // 创建 ThreadPoolTaskExecutor 对象
            return builder.build();
        }

    }

    @Configuration
    public static class ExecutorTwoConfiguration
    {

        @Bean(name = EXECUTOR_TWO_BEAN_NAME + "-properties")
        @ConfigurationProperties(prefix = "spring.task.execution-two")
        // 读取 spring.task.execution-two 配置到 TaskExecutionProperties 对象
        public TaskExecutionProperties taskExecutionProperties()
        {
            return new TaskExecutionProperties();
        }

        @Bean(name = EXECUTOR_TWO_BEAN_NAME)
        public ThreadPoolTaskExecutor threadPoolTaskExecutor()
        {
            // 创建 TaskExecutorBuilder 对象
            TaskExecutorBuilder builder = createTskExecutorBuilder(this.taskExecutionProperties());
            // 创建 ThreadPoolTaskExecutor 对象
            return builder.build();
        }
    }

    private static TaskExecutorBuilder createTskExecutorBuilder(TaskExecutionProperties properties)
    {
        // Pool 属性
        TaskExecutionProperties.Pool pool = properties.getPool();
        TaskExecutorBuilder builder = new TaskExecutorBuilder();
        builder = builder.queueCapacity(pool.getQueueCapacity());
        builder = builder.corePoolSize(pool.getCoreSize());
        builder = builder.maxPoolSize(pool.getMaxSize());
        builder = builder.allowCoreThreadTimeOut(pool.isAllowCoreThreadTimeout());
        builder = builder.keepAlive(pool.getKeepAlive());
        // Shutdown 属性
        TaskExecutionProperties.Shutdown shutdown = properties.getShutdown();
        builder = builder.awaitTermination(shutdown.isAwaitTermination());
        builder = builder.awaitTerminationPeriod(shutdown.getAwaitTerminationPeriod());
        // 其它基本属性
        builder = builder.threadNamePrefix(properties.getThreadNamePrefix());
//        builder = builder.customizers(taskExecutorCustomizers.orderedStream()::iterator);
//        builder = builder.taskDecorator(taskDecorator.getIfUnique());
        return builder;
    }

}

复制代码@Service
public class DemoService
{

    private Logger logger = LoggerFactory.getLogger(getClass());

    @Async(AsyncConfig.EXECUTOR_ONE_BEAN_NAME)
    public Integer execute01()
    {
        logger.info("[execute01]");
        return 1;
    }

    @Async(AsyncConfig.EXECUTOR_TWO_BEAN_NAME)
    public Integer execute02()
    {
        logger.info("[execute02]");
        return 2;
    }

}

复制代码@RunWith(SpringRunner.class)
@SpringBootTest(classes = Application.class)
public class DemoServiceTest
{

    @Autowired
    private DemoService demoService;

    @Test
    public void testExecute() throws InterruptedException
    {
        demoService.execute01();
        demoService.execute02();

        // sleep 1 秒，保证异步调用的执行
        Thread.sleep(1000);
    }

}

复制代码2020-06-08 15:38:28.846  INFO 12020 --- [     task-one-1] c.i.s.l.asynctask.service.DemoService    : [execute01]
2020-06-08 15:38:28.846  INFO 12020 --- [     task-two-1] c.i.s.l.asynctask.service.DemoService    : [execute02]

复制代码@Configuration
public class SecurityConfig extends WebSecurityConfigurerAdapter {
    @Bean
    PasswordEncoder passwordEncoder() {
        return new BCryptPasswordEncoder();
    }

    @Override
    @Bean
    public AuthenticationManager authenticationManagerBean() throws Exception {
        return super.authenticationManagerBean();
    }

    @Override
    protected void configure(AuthenticationManagerBuilder auth) throws Exception {
        auth.inMemoryAuthentication()
                .withUser("sang")
                .password(passwordEncoder().encode("123"))
                .roles("admin")
                .and()
                .withUser("javaboy")
                .password(passwordEncoder().encode("123"))
                .roles("user");
    }

    @Override
    protected void configure(HttpSecurity http) throws Exception {
        http.csrf().disable().formLogin();
    }
}

复制代码@Configuration
public class AccessTokenConfig {
    @Bean
    TokenStore tokenStore() {
        return new JwtTokenStore(jwtAccessTokenConverter());
    }

    @Bean
    JwtAccessTokenConverter jwtAccessTokenConverter() {
        JwtAccessTokenConverter converter = new JwtAccessTokenConverter();
        converter.setSigningKey("javaboy");
        return converter;
    }
}

复制代码@EnableAuthorizationServer
@Configuration
public class AuthorizationServer extends AuthorizationServerConfigurerAdapter {
    @Autowired
    TokenStore tokenStore;
    @Autowired
    ClientDetailsService clientDetailsService;
    @Autowired
    AuthenticationManager authenticationManager;
    @Autowired
    PasswordEncoder passwordEncoder;
    @Autowired
    JwtAccessTokenConverter jwtAccessTokenConverter;

    @Bean
    AuthorizationServerTokenServices tokenServices() {
        DefaultTokenServices services = new DefaultTokenServices();
        services.setClientDetailsService(clientDetailsService);
        services.setSupportRefreshToken(true);
        services.setTokenStore(tokenStore);
        services.setAccessTokenValiditySeconds(60 * 60 * 24 * 2);
        services.setRefreshTokenValiditySeconds(60 * 60 * 24 * 7);
        TokenEnhancerChain tokenEnhancerChain = new TokenEnhancerChain();
        tokenEnhancerChain.setTokenEnhancers(Arrays.asList(jwtAccessTokenConverter));
        services.setTokenEnhancer(tokenEnhancerChain);
        return services;
    }

    @Override
    public void configure(AuthorizationServerSecurityConfigurer security) throws Exception {
        security.allowFormAuthenticationForClients();
    }

    @Override
    public void configure(ClientDetailsServiceConfigurer clients) throws Exception {
        clients.inMemory()
                .withClient("javaboy")
                .secret(passwordEncoder.encode("123"))
                .resourceIds("res1")
                .authorizedGrantTypes("password", "refresh_token")
                .scopes("all")
                .redirectUris("http://localhost:8082/index.html");
    }

    @Override
    public void configure(AuthorizationServerEndpointsConfigurer endpoints) throws Exception {
        endpoints
                .authenticationManager(authenticationManager)
                .tokenServices(tokenServices());
    }
}

复制代码@Configuration
@EnableResourceServer
public class ResourceServerConfig extends ResourceServerConfigurerAdapter {
    @Autowired
    TokenStore tokenStore;

    @Override
    public void configure(ResourceServerSecurityConfigurer resources) throws Exception {
        resources.resourceId("res1").tokenStore(tokenStore);
    }

    @Override
    public void configure(HttpSecurity http) throws Exception {
        http.authorizeRequests()
                .antMatchers("/admin/**").hasRole("admin")
                .anyRequest().authenticated();
    }
}

复制代码@RestController
public class HelloController {
    @GetMapping("/hello")
    public String hello() {
        return "hello";
    }
    @GetMapping("/admin/hello")
    public String admin() {
        return "admin";
    }
}

复制代码 <dependency>
     <groupId>io.springfox</groupId>
     <artifactId>springfox-swagger2</artifactId>
     <version>2.9.2</version>
 </dependency>
 <dependency>
     <groupId>io.springfox</groupId>
     <artifactId>springfox-swagger-ui</artifactId>
     <version>2.9.2</version>
 </dependency>