Flink-异步算子AsyncFunction利用 - ToB企服应用市场:ToB评测及商务社交产业平台

package app;
import com.alibaba.fastjson.JSONObject;
import functions.MyAsyncFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.concurrent.TimeUnit;
public class AsyncTestJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
SingleOutputStreamOperator<JSONObject> stream = env.socketTextStream("localhost", 8888)
.map((MapFunction<String, JSONObject>) JSONObject::parseObject);
AsyncDataStream.orderedWait(stream,
//使用异步算子
new MyAsyncFunction(50),
//超时时间
30000,
//超时时间单位
TimeUnit.MILLISECONDS,
//capacity为异步请求数参数。建议:该参数应小于等于线程池、连接池的最大连接数
50)
.print();
env.execute();
}
}

复制代码

package functions;
import com.alibaba.fastjson.JSONObject;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.*;
public class MyAsyncFunction extends RichAsyncFunction<JSONObject, String> {
private final int maxConnTotal;
private transient ExecutorService executorService;
private HikariDataSource phoenixDataSource;
private final static Logger logger = LoggerFactory.getLogger(AsyncUpsertHBase.class);
public MyAsyncFunction(Integer maxConnTotal) {
this.maxConnTotal = maxConnTotal;
}
@Override
public void open(Configuration parameters) throws Exception {
executorService = Executors.newFixedThreadPool(maxConnTotal);
phoenixDataSource = getPhoenixDataSource();
}
@Override
public void close() throws Exception {
executorService.shutdown();
phoenixDataSource.close();
}
@Override
public void asyncInvoke(JSONObject json, ResultFuture<String> resultFuture) {
//使用线程池实现异步提交请求
Future<String> future = executorService.submit(() -> upsertHBase(json));
CompletableFuture.supplyAsync(() -> {
try {
return future.get();
} catch (InterruptedException | ExecutionException e) {
return null;
}
}).thenAccept((String dbResult) -> resultFuture.complete(Collections.singleton(dbResult)));
}
/**
* 标签写入HBase
*
* @param json 标签数据
* @return 标签数据字符串
* @throws SQLException SQLException
*/
private String upsertHBase(JSONObject json) throws SQLException {
Long id = json.getLong("id");
String key = json.getString("key");
Integer value = json.getInteger("data");
String SQL = "UPSERT INTO TEST_TABLE (ID," + key + ") VALUES (" + id + "," + value + ")";
try (
Connection conn = phoenixDataSource.getConnection();
PreparedStatement ps = conn.prepareStatement(SQL)
) {
ps.executeUpdate();
conn.commit();
return json.toJSONString();
} catch (Exception e) {
logger.error(e.getMessage(), e);
throw e;
}
}
private HikariDataSource getPhoenixDataSource() {
HikariConfig config = new HikariConfig();
config.setDriverClassName("org.apache.phoenix.jdbc.PhoenixDriver");
config.setJdbcUrl("jdbc:phoenix:10.xxx.xxx.xxx:2181");
// 最大活跃连接数
config.setMaximumPoolSize(50);
//最小空闲连接数
config.setMinimumIdle(10);
//连接超时时间，单位毫秒
config.setConnectionTimeout(30000);
//空闲连接最大存活时间，单位毫秒
config.setIdleTimeout(5 * 60 * 1000);
//初始化失败时的超时时间，单位秒
config.setInitializationFailTimeout(1);
//池名称
config.setPoolName("PhoenixHikariCP");
// 允许池在达到最大大小时增长，如果当前所有连接都在使用中，并且执行者正在等待一个连接
config.setAllowPoolSuspension(true);
Properties properties = new Properties();
properties.setProperty("phoenix.schema.mapSystemTablesToNamespace", "true");
properties.setProperty("phoenix.schema.isNamespaceMappingEnabled", "true");
properties.setProperty("phoenix.query.timeoutMs", "1200000");
properties.setProperty("hbase.rpc.timeout", "1200000");
properties.setProperty("hbase.client.scanner.timeout.period", "1200000");
config.setDataSourceProperties(properties);
return new HikariDataSource(config);
}
}

复制代码