Flink异步请求高德地图获取位置信息

1、使用异步方法，异步请求高德地图获取位置信息

2、代码实现

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

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

        AsyncDataStream.unorderedWait(data,new AstncGaodeMap(),30000, TimeUnit.MILLISECONDS,10).print();
        env.execute();
    }
}

/**
 * <!-- 高效的异步HttpClient依赖 -->
 *                 <dependency>
 *                    <groupId>org.apache.httpcomponents</groupId>
 *                    <artifactId>httpasyncclient</artifactId>
 *                    <version>4.1.4</version>
 *                </dependency>
 */
class AstncGaodeMap extends RichAsyncFunction<String,OrderBean>
{
    private CloseableHttpAsyncClient httpAsyncClient;
    private String key = "开发平台申请";
    private String province;
    private String city;
    @Override
    public void open(Configuration parameters) throws Exception {
        //创建异步查询的HTTPClient
        //创建一个异步的HttpClient连接池
        //初始化异步的HttpClient
        RequestConfig requestConfig = RequestConfig.custom()
                .setSocketTimeout(3000)
                .setConnectTimeout(3000)
                .build();

         httpAsyncClient = HttpAsyncClients.custom()
                .setMaxConnTotal(10)
                .setDefaultRequestConfig(requestConfig)
                .build();

        httpAsyncClient.start();
    }

    @Override
    public void asyncInvoke(String input, ResultFuture<OrderBean> resultFuture) throws Exception {
        OrderBean orderBean = JSON.parseObject(input, OrderBean.class);
        String latitude = orderBean.getLatitude();
        String longitude = orderBean.getLongitude();
        //创建异步查询,使用restAPI
        HttpGet httpGet = new HttpGet("https://restapi.amap.com/v3/geocode/regeo?&location=" + longitude + "," + latitude + "&key=" + key);

        //执行查询，返回future
        Future<HttpResponse> future = httpAsyncClient.execute(httpGet, null);

        //从future中获取数据
        CompletableFuture.supplyAsync(new Supplier<OrderBean>() {
            @Override
            public OrderBean get() {
                try {
                    HttpResponse httpResponse = future.get();
                    if(httpResponse.getStatusLine().getStatusCode() == 200)
                    {
                        //获取请求的json字符串
                        String result = EntityUtils.toString(httpResponse.getEntity());
                        System.out.println(result);
                        //转成json对象
                        JSONObject jsonObject = JSON.parseObject(result);
                        //获取位置信息
                        JSONObject regeocode = jsonObject.getJSONObject("regeocode");
                        if(regeocode != null && !regeocode.isEmpty())
                        {
                            JSONObject address = regeocode.getJSONObject("addressComponent");
                            province = address.getString("province");
                            city = address.getString("city");
                        }
                    }
                    orderBean.setProvince(province);
                    orderBean.setCity(city);
                    return orderBean;
                } catch (Exception e) {
                    e.printStackTrace();
                    return null;
                }
            }
        }).thenAccept((OrderBean result)->{
            resultFuture.complete(Collections.singleton(result));
        });

    }
    @Override
    public void close() throws Exception {
        httpAsyncClient.close();
    }
}

3、测试数据

• 输入结果

• {“oid”: “o1000”, “cid”: “c10”, “money”: 99.99, “longitude”: 116.413467, “latitude”: 39.908072}

• 输出结果

• {“status”:“1”,“regeocode”:{“addressComponent”:{“city”:[],“province”:“北京市”,“adcode”:“110101”,“district”:“东城区”,“towncode”:“110101001000”,“streetNumber”:{“number”:“4号”,“location”:“116.413426,39.907697”,“direction”:“南”,“distance”:“41.8452”,“street”:“东长安街”},“country”:“中国”,“township”:“东华门街道”,“businessAreas”:[{“location”:“116.412031,39.913505”,“name”:“王府井”,“id”:“110101”},{“location”:“116.416804,39.913479”,“name”:“东单”,“id”:“110101”},{“location”:“116.425171,39.929561”,“name”:“东四”,“id”:“110101”}],“building”:{“name”:[],“type”:[]},“neighborhood”:{“name”:[],“type”:[]},“citycode”:“010”},“formatted_address”:“北京市东城区东华门街道东长安街”},“info”:“OK”,“infocode”:“10000”}

• OrderBean(oid=o1000, cid=c10, money=99.99, longitude=116.413467, latitude=39.908072, province=北京市, city=[])