提出问题

问题一

flink 收到一条数据后，根据该条数据，要对 redis，hbase 进行查询、修改、删除等操作

问题二

flink 收到一条数据后，由于该数据不完整，要调 api 填充缺失的信息

分析-使用 MapFunction 算子

一般在 MapFunction 这类算子中进行 操作数据库、调用 api 的操作，是以 同步方式 操作的，即: 前一个数据处理完后，才处理下一个数据

缺点： 前一个数据，要到外部数据库中（hbase、MySQL）进行增删改查，此时可能会发生IO阻塞，等待请求返回，这种同步交互的方式往往在网络等待上就耗费了大量时间 。前一个数据处理后，才会继续处理下一个数据，性能低下

例子

package hbase;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.concurrent.TimeUnit;

public class Std1FlatMap {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<String> dss = env.fromElements(
                "1李雷 2lucy",
                "3韩梅梅 4lili",
                "5李雷 6韩梅梅");
        // map解析json数据
        SingleOutputStreamOperator<Tuple2<String,Integer>> mapSOSO = dss.flatMap(new FlatMapFunImpl());

        SingleOutputStreamOperator<Object> soso = mapSOSO.map(new MapFunction<Tuple2<String, Integer>, Object>() {
            @Override
            public Object map(Tuple2<String, Integer> value) throws Exception {
                Thread.sleep(500);
                System.out.println(String.format("模拟从数据库查询数据消耗的时间::%s ms",500 ));
                return value;
            }
        });
        soso.print();
        env.execute();
    }
    public static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] arr = value.split(" ");
            for(String item : arr){
                out.collect(Tuple2.of(item,1));
            }
        }
    }
}

问题与解决

与外部系统交互时，网络延迟成为系统瓶颈，前一个数据处理慢，后面的数据都处于等待

使用 Flink Async I/O 异步处理

Async I/O

Flink 在 1.2 中引入了Async I/O，在 异步模式 下，将IO操作异步化，并行发送多个请求，哪个先执行完就先返回，不需要阻塞等待，大大提高了流处理效率。

注意：要求外部系统客户端支持异步IO，否则需要 自定义实现异步IO

AsyncDataStream 类

有两个静态方法：

• unorderedWait，不关注顺序，哪个任务先执行完，就先返回数据

• orderedWait，关注顺序，输出的顺序和输入的顺序是一样的

unorderedWait() 方法

表示数据 不需要关注顺序，异步请求执行结束后，立即输出结果
假设：前一个数据，处理时间需要 6秒，后一个数据，处理时间需要 3秒，执行结果：由于后一个数据处理时间较短，所以 后一个数据的结果会先输出

参数解释：

• 第一个参数：上游 datastream 流

• 第二个参数：自己实现的类，需要继承 RichAsyncFunction 类或者 实现 AsyncFunction 接口。一般继承 RichAsyncFunction，有生命周期方法

• 第三个参数：是最长等待时间

• 第四个参数：是 TimeUnit 可以传入TimeUnit.SECONDS 这种

• 最后一个参数：表示同时最多有多少个异步请求在处理，异步IO的方式会导致更高的吞吐量，但是对于实时应用来说该操作也是一个瓶颈。限制并发请求数，算子不会积压过多的未处理请求，但是一旦超过容量的显示会触发背压。
  该参数可以不配置，但是默认是 100

AsyncFunction 类

有 asyncInvoke() 方法，用于异步执行，形参：

• 第一个形参是传入的数据
• 第二个形参用于收集返回的数据

RichAsyncFunction 类

有以下方法：

• open() 方法：初始化方法，被调用之前， open() 会被调用，只会执行一次。如：连接数据库

• asyncInvoke() 方法，用于异步执行，形参：

  • 第一个形参是传入的数据
  • 第二个形参用于收集返回的数据

• close() 方法：是生命周期中的 最后一个调用的方法，做一些清理工作。如：关闭数据库

例子-失败

本例子只是简单使用，虽然使用异步的 AsyncFunction 类，但客户端不支持异步IO，也没有自定义实现异步IO，所以执行时还是 同步的

package hbase;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Random;
import java.util.concurrent.TimeUnit;

public class Std2AsyncDataStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 测试数据
        DataStreamSource<String> dss = env.fromElements(
                "1李雷 2lucy",
                "3韩梅梅 4lili",
                "5李雷 6韩梅梅");
        // map解析json数据
        SingleOutputStreamOperator<Tuple2<String,Integer>> mapSOSO = dss.flatMap(new FlatMapFunImpl());

        SingleOutputStreamOperator<Object> soso = AsyncDataStream.unorderedWait(mapSOSO, new AsyncFunction<Tuple2<String, Integer>, Object>() {

            @Override
            public void asyncInvoke(Tuple2<String, Integer> input, ResultFuture<Object> resultFuture) throws Exception {
                int delay = new Random().nextInt(10) * 100;
                System.out.println(String.format("模拟从数据库查询数据消耗的时间::%s ms",delay ));
                Thread.sleep(delay);
                // 通过下面方法收集数据
                resultFuture.complete(Collections.singleton(input));
            }
        }, 60, TimeUnit.SECONDS);
        soso.print();
        env.execute();
    }
    public static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] arr = value.split(" ");
            for(String item : arr){
                out.collect(Tuple2.of(item,1));
            }
        }
    }
}

例子-成功

为了实现异步IO，本例自定义实现异步IO，执行代码必须放在线程中，一般搭配 ThreadPoolExecutor 线程池使用

需要使用 RichAsyncFunction 类

package hbase;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Std3AsyncDataStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 模拟数据
        DataStreamSource<String> dss = env.fromElements(
                "1李雷 2lucy",
                "3韩梅梅 4lili",
                "5李雷 6韩梅梅");
        // map解析json数据
        SingleOutputStreamOperator<Tuple2<String,Integer>> mapSOSO = dss.flatMap(new FlatMapFunImpl());

        SingleOutputStreamOperator<Object> soso = AsyncDataStream.unorderedWait(mapSOSO, new RichAsyncFunction<Tuple2<String, Integer>, Object>() {
            // 创建线程池
            private ThreadPoolExecutor threadPoolExecutor;
            @Override
            public void open(Configuration parameters) throws Exception {
                //初始化线程池
                threadPoolExecutor =  new ThreadPoolExecutor(
                        10,
                        10,
                        0,
                        TimeUnit.SECONDS,
                        new ArrayBlockingQueue(1000));
            }

            /**
             * 每来一条数据，就执行一次该方法
             * @param input 输入数据
             * @param resultFuture 通过该对象收集数据
             * @throws Exception
             */
            @Override
            public void asyncInvoke(Tuple2<String, Integer> input, ResultFuture<Object> resultFuture) throws Exception {
                int delay = new Random().nextInt(10) * 100;
                threadPoolExecutor.submit(new Runnable() {
                    @Override
                    public void run() {

                        try {
                            System.out.println(String.format("模拟从数据库查询数据消耗的时间::%s ms，数据：%s",delay,input ));
                            Thread.sleep(delay);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        // 通过下面方法收集数据
                        resultFuture.complete(Collections.singleton(input));
                    }
                });
            }
        }, 30, TimeUnit.SECONDS);
        soso.print();
        env.execute();
    }
    public static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] arr = value.split(" ");
            for(String item : arr){
                out.collect(Tuple2.of(item,1));
            }
        }
    }
}

执行结果：

超时处理

当超时没有返回结果时，会自动调用 AsyncFunction 接口的方法：

default void timeout(IN input, ResultFuture<OUT> resultFuture) throws Exception

例子

package hbase;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Std3AsyncDataStream_超时处理 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 模拟数据
        DataStreamSource<String> dss = env.fromElements(
                "1李雷 2lucy",
                "3韩梅梅 4lili",
                "5李雷 6韩梅梅");
        // map解析json数据
        SingleOutputStreamOperator<Tuple2<String,Integer>> mapSOSO = dss.flatMap(new FlatMapFunImpl());

        SingleOutputStreamOperator<Object> soso = AsyncDataStream.unorderedWait(
                mapSOSO,
                new MyRichAsyncFunction(),
                5, // 超时时间，如果任务超时5s还没有返回，就执行 timeout() 方法
                TimeUnit.SECONDS // 超时单位
        );
        soso.print();
        env.execute();
    }

    static class MyRichAsyncFunction extends RichAsyncFunction<Tuple2<String, Integer>, Object> {
        // 创建线程池
        private ThreadPoolExecutor threadPoolExecutor;
        @Override
        public void open(Configuration parameters) throws Exception {
            //初始化线程池
            threadPoolExecutor =  new ThreadPoolExecutor(
                    10,
                    10,
                    0,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue(1000));
        }

        /**
         * 每来一条数据，就执行一次该方法
         * @param input 输入数据
         * @param resultFuture 通过该对象收集数据
         * @throws Exception
         */
        @Override
        public void asyncInvoke(Tuple2<String, Integer> input, ResultFuture<Object> resultFuture) throws Exception {
            int delay = new Random().nextInt(10) * 1000;
            threadPoolExecutor.submit(new Runnable() {
                @Override
                public void run() {

                    try {
                        System.out.println(String.format("模拟从数据库查询数据消耗的时间::%s ms，数据：%s",delay,input ));
                        Thread.sleep(delay);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                    // 通过下面方法收集数据
                    resultFuture.complete(Collections.singleton(input));
                }
            });
        }

        @Override
        public void timeout(Tuple2<String, Integer> input, ResultFuture<Object> resultFuture) throws Exception {
            resultFuture.complete(Collections.singleton(String.format("数据：%s，查询超时！", input)));
        }
    }

    static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] arr = value.split(" ");
            for(String item : arr){
                out.collect(Tuple2.of(item,1));
            }
        }
    }
}

orderedWait() 方法

表示数据需要关注顺序，输出的顺序和输入的顺序是一样的

如上图，尽管 P4 的结果先返回，但并不会发送，只有 P1 （队首）的结果返回了，才会将P1（队首）的结果发送

实现原理：将 P4 结果缓存在队列中，在此之前的数据 P1、P2、P3全部完成并输出后，才输出 P4 的结果

缺点：

• 带来了一些额外的延迟
• 由于数据保存在 checkpoint 内的时间会更长, 故 checkpoint 负载会增加

形参：传入参数与 unorderedWait() 相同

例子

package hbase;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.flink.util.Collector;

import java.util.Collections;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class Std5AsyncDataStream_order {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 模拟数据
        DataStreamSource<String> dss = env.fromElements(
                "1李雷 2lucy",
                "3韩梅梅 4lili",
                "5李雷 6韩梅梅");
        // map解析json数据
        SingleOutputStreamOperator<Tuple2<String,Integer>> mapSOSO = dss.flatMap(new FlatMapFunImpl());

        SingleOutputStreamOperator<Object> soso = AsyncDataStream.orderedWait(mapSOSO, new RichAsyncFunction<Tuple2<String, Integer>, Object>() {
            // 创建线程池
            private ThreadPoolExecutor threadPoolExecutor;
            @Override
            public void open(Configuration parameters) throws Exception {
                //初始化线程池
                threadPoolExecutor =  new ThreadPoolExecutor(
                        10,
                        10,
                        0,
                        TimeUnit.SECONDS,
                        new ArrayBlockingQueue(1000));
            }

            /**
             * 每来一条数据，就执行一次该方法
             * @param input 输入数据
             * @param resultFuture 通过该对象收集数据
             * @throws Exception
             */
            @Override
            public void asyncInvoke(Tuple2<String, Integer> input, ResultFuture<Object> resultFuture) throws Exception {
                int delay = new Random().nextInt(10) * 100;
                threadPoolExecutor.submit(new Runnable() {
                    @Override
                    public void run() {

                        try {
                            System.out.println(String.format("模拟从数据库查询数据消耗的时间::%s ms，数据：%s",delay,input ));
                            Thread.sleep(delay);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        // 通过下面方法收集数据
                        resultFuture.complete(Collections.singleton(input));
                    }
                });
            }
        }, 30, TimeUnit.SECONDS);
        soso.print();
        env.execute();
    }
    public static class FlatMapFunImpl implements FlatMapFunction<String, Tuple2<String,Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] arr = value.split(" ");
            for(String item : arr){
                out.collect(Tuple2.of(item,1));
            }
        }
    }
}

执行结果：

方法选择

使用 unorderedWait 还是 orderedWait，主要看当前业务是否要保证顺序一致，如果不关注顺序，就选择 unorderedWait，如果要求顺序不能乱，就选择 orderedWait

真实案例

查询外部存储系统（MySQL、redis、Hbase）的案例：

• flink 维表关联、翻译字典-热存储维表-使用AsyncDataStream、RichAsyncFunction 进行异步处理（查询MySQL）

• flink 维表关联、翻译字典-热存储维表-使用AsyncDataStream、RichAsyncFunction 查询Hbase

参考：
https://blog.csdn.net/a724952091/article/details/108580882
https://zhuanlan.zhihu.com/p/364868336
https://zhuanlan.zhihu.com/p/268898593

原文出处：/show_1IX4cEJSQSey.html