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

Posted 2022-05-02 猫猫爱吃小鱼粮

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=[])