60、Flink 的异步 IO 算子使用异步 Http 客户端查高德地图

1、概述

  Http 异步客户端
  设置：并行度=2，capacity=2，HttpMaxConn=2，client 为静态
  输入：同时发起4条查询
  输出：间隔10秒，同时返回4条数据

  JDBC 线程池+链接池
  设置：并行度=2，capacity=2，HttpMaxConn=2，client 为静态
  输入：同时发起4条查询
  输出：间隔10秒，先返回两条数据，间隔10秒，再返回两条数据
1
2
3
4
5
6
7
8
9

2、代码示例

package com.xu.flink.datastream.day11;

import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
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.apache.flink.util.concurrent.Executors;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.concurrent.TimeUnit;

public class _06_HttpAsyncQueryGaoDe {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

        AsyncDataStream.unorderedWait(lines, new AsyncHttpQueryFunction(), 20, TimeUnit.SECONDS, 2).print();

        env.execute();
    }
}

/**
 * Http 异步客户端
 * 设置：并行度=2，capacity=2，HttpMaxConn=2，client 为静态
 * 输入：同时发起4条查询
 * 输出：间隔10秒，同时返回4条数据
 * <p>
 * JDBC 线程池+链接池
 * 设置：并行度=2，capacity=2，HttpMaxConn=2，client 为静态
 * 输入：同时发起4条查询
 * 输出：间隔10秒，先返回两条数据，间隔10秒，再返回两条数据
 */
class AsyncHttpQueryFunction extends RichAsyncFunction<String, _06_OrderBean> {

    private static final String key = "***";
    private static CloseableHttpAsyncClient httpClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        //创建异步 HttpClient 连接池，并初始化异步的 HttpClient
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(3000)
                .setConnectTimeout(3000)
                .build();

        httpClient = HttpAsyncClients.custom()
                .setMaxConnTotal(2)
                .setDefaultRequestConfig(requestConfig)
                .build();

        System.out.println("open 方法被调用");
        System.out.println("httpClient=>" + httpClient.hashCode());

        httpClient.start();
    }

    @Override
    public void asyncInvoke(String line, ResultFuture<_06_OrderBean> resultFuture) throws Exception {

        try {
            // 1、解析JSON，获取经纬度信息
            _06_OrderBean orderBean = JSON.parseObject(line, _06_OrderBean.class);
            double longitude = orderBean.getLongitude();
            double latitude = orderBean.getLatitude();

            // 创建 httpGet 请求
            HttpGet httpGet = new HttpGet("https://restapi.amap.com/v3/geocode/regeo?&location=" + longitude + "," + latitude + "&key=" + key);

            // 2、通过 httpClient 提交异步请求，获取 future 对象
            // callback 是回调函数（也可通过回调函数拿结果）

            // 注意：此处使用 task 线程，如果此处是异步提交，则不会阻塞；如果此处是同步提交，则会阻塞；
            Future<HttpResponse> future = httpClient.execute(httpGet, null);

            // 3、从成功的 Future 中取数据，返回 orderBean
            // 使用 Executors.directExecutor() 获取返回结果

            // 注意：此处为异步获取返回结果，会使用单独的线程池，即使用 get() 方法，也不会阻塞 task 线程
            CompletableFuture.supplyAsync(new Supplier<_06_OrderBean>() {
                        @Override
                        public _06_OrderBean get() {
                            try {
                                HttpResponse response = future.get();

                                String province = null;
                                String city = null;
                                String district = null;


                                if (response.getStatusLine().getStatusCode() == 200) {
                                    //拿出响应的实例对象
                                    HttpEntity entity = response.getEntity();

                                    JSONObject jsonObject = JSON.parseObject(EntityUtils.toString(entity));

                                    JSONObject regeocodeObject = jsonObject.getJSONObject("regeocode");

                                    if (regeocodeObject != null && !regeocodeObject.isEmpty()) {
                                        JSONObject addObject = regeocodeObject.getJSONObject("addressComponent");
                                        district = addObject.getString("district");
                                        city = addObject.getString("city");
                                        province = addObject.getString("province");
                                    }
                                }
                                orderBean.setProvince(province);
                                orderBean.setCity(city);
                                orderBean.setDistrict(district);

                                return orderBean;
                            } catch (Exception e) {
                                return null;
                            }
                        }
                    }, Executors.directExecutor())
                    .thenAccept(new Consumer<_06_OrderBean>() {
                        @Override
                        public void accept(_06_OrderBean resultOrderBean) {
                            resultFuture.complete(Collections.singleton(resultOrderBean));
                        }
                    });

        } catch (Exception e) {
            resultFuture.complete(Collections.singleton(null));
        }
    }

    @Override
    public void timeout(String input, ResultFuture<_06_OrderBean> resultFuture) throws Exception {
        resultFuture.completeExceptionally(new RuntimeException(input + "=查询超时"));
    }

    @Override
    public void close() throws Exception {
        httpClient.close();
    }
}
-----------------------------------------------------------------
@Data
@NoArgsConstructor
@AllArgsConstructor
public class _06_OrderBean {
    private String oid;
    private String uid;
    private double money;
    private double longitude;
    private double latitude;
    private String province;
    private String city;
    private String district;

    @Override
    public String toString() {
        return "OrderBean{" +
                "oid='" + oid + '\'' +
                ", uid='" + uid + '\'' +
                ", money=" + money +
                ", longitude=" + longitude +
                ", latitude=" + latitude +
                ", province='" + province + '\'' +
                ", city='" + city + '\'' +
                ", district='" + district + '\'' +
                '}';
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183

3、测试用例

{"oid":"o001","uid":"u001","money":99.99,"longitude":115.690417, "latitude":36.239344}
1