Flink批量写入mysql

<insert id="insertBatch" parameterType="java.util.List">
    insert into sku_info (sku,num) values
    <foreach collection="list" item="info" index="index" separator=",">
        (#{info.skuId}, #{info.num})
    </foreach>
</insert>

void insertBatch(List<Info> infos);

package com.jd.xq;
import com.google.common.collect.Lists;
import com.jd.xq.mapper.LineMapper;
import com.jd.xq.model.SkuInfo;
import com.jd.xq.sink.SinkSku;
import com.jd.xq.source.TestSource;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;
/**
 * @author
 * @Date 2019-08-16 15:45
 **/
public class StartJob {
    public static void main(String[] args) {
        try {
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.enableCheckpointing(5000); // 非常关键，一定要设置启动检查点！！
           env.getConfig().setGlobalJobParameters(loadConfig("config.properties"));
            DataStream<String> text = env.addSource(new TestSource());
            DataStream stream = text.flatMap(new LineMapper());
            stream.countWindowAll(3).apply(new AllWindowFunction<SkuInfo, List<SkuInfo>, Window>() {
                @Override
                public void apply(Window window, Iterable<SkuInfo> iterable, Collector<List<SkuInfo>> collector) throws Exception {
                    ArrayList<SkuInfo> skuInfos = Lists.newArrayList(iterable);
                    if (skuInfos.size() > 0) {
                        collector.collect(skuInfos);
                    }
                }
            }).addSink(new SinkSku());
            env.execute("WordCount from Kafka data");
        } catch (Exception e) {
            System.out.println(e);
        }
    }
    public static ParameterTool loadConfig(String configFileName) throws Exception {
        try (InputStream is = StartJob.class.getClassLoader().getResourceAsStream(configFileName)) {
            return ParameterTool.fromPropertiesFile(is);
        }
    }
}

public class TestSource extends RichSourceFunction<String> {
    private volatile boolean isRunning = true;
    @Override
    public void open(Configuration parameters) throws Exception {
    }
    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        Random random = new Random();
        while (isRunning) {
            Thread.sleep((getRuntimeContext().getIndexOfThisSubtask() + 1) * 1000);
            //String key = "类别" + (char) ('A' + random.nextInt(3));
            int sku = random.nextInt(10);
            int value = random.nextInt(10) + 1;
            String str = sku + "," + value;
            //System.out.println(str);
            sourceContext.collect(str);
        }
    }
    @Override
    public void cancel() {
        isRunning = false;
    }
}

package com.jd.xq.mapper;
import com.alibaba.fastjson.JSONObject;
import com.jd.xq.model.Info;
import com.jd.xq.model.SkuInfo;
import com.jd.xq.model.TestModel;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
/**
 * @author duanxiaoqiu
 * @Date 2019-09-10 20:50
 **/
public class LineMapper extends RichFlatMapFunction<String, SkuInfo> {
    @Override
    public void open(Configuration parameters) {
        ExecutionConfig executionConfig = getRuntimeContext().getExecutionConfig();
        ParameterTool params = (ParameterTool) executionConfig.getGlobalJobParameters();
        //System.out.println(params.get("xq.name"));
    }
    public void flatMap(String s, Collector<SkuInfo> collector) {
        try {
            String[] arr = s.split(",");
            SkuInfo model = new SkuInfo();
            model.setSkuId(arr[0]);
            model.setNum(Integer.parseInt(arr[1]));
            //incr(model.getSku(), model.getNum());
            System.out.println(JSONObject.toJSONString(model));
            collector.collect(model);
        } catch (Exception e) {
            System.out.println();
        }
    }
}

public class SinkSku extends RichSinkFunction<List<SkuInfo>> {
    private SkuMapper skuMapper;
    @Override
    public void open(Configuration parameters) {
        try {
            skuMapper = SkuDao.getInstance();
        } catch (Exception e) {
            System.out.println("====open====");
        }
    }
    @Override
    public void invoke(List<SkuInfo> value, Context context) {
        try {
            System.out.println(JSONObject.toJSONString(value));
            skuMapper.insertBatch(value);
        } catch (Exception e) {
            System.out.println("====invoke e:" + e);
        }
    }
}

stream.countWindowAll(3).apply(new AllWindowFunction<SkuInfo, List<SkuInfo>, Window>() {
    @Override
    public void apply(Window window, Iterable<SkuInfo> iterable, Collector<List<SkuInfo>> collector) throws Exception {
        ArrayList<SkuInfo> skuInfos = Lists.newArrayList(iterable);
        if (skuInfos.size() > 0) {
            collector.collect(skuInfos);
        }
    }
}).addSink(new SinkSku());

stream.timeWindowAll(Time.seconds(2)).apply(new AllWindowFunction<SkuInfo, List<SkuInfo>, TimeWindow>() {
    @Override
    public void apply(TimeWindow window, Iterable<SkuInfo> iterable, Collector<List<SkuInfo>> collector) {
        ArrayList<SkuInfo> skuInfos = null;
        try {
            skuInfos = Lists.newArrayList(iterable);
            if (skuInfos.size() > 0) {
                collector.collect(skuInfos);
            }
        } catch (Exception e) {
            System.out.println("==window e:" + e);
        }
    }
}).addSink(new SinkSku());