httpclient 池化，异步化

桃扇骨 2023-06-12 11:05 38阅读 0赞

import lombok.extern.slf4j.Slf4j;
    import org.apache.http.client.config.RequestConfig;
    import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
    import org.apache.http.impl.nio.client.HttpAsyncClients;
    import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
    import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
    import org.apache.http.impl.nio.reactor.IOReactorConfig;
    import org.apache.http.nio.reactor.ConnectingIOReactor;
    import org.apache.http.nio.reactor.IOReactorException;
    import org.springframework.util.Assert;
    
    @Slf4j
    public class HttpAsyncClient {
    
        // 异步httpclient
        private volatile static CloseableHttpAsyncClient asyncHttpClient;
    
        private static final Object LOCK = new Object();
    
        static {
            asyncHttpClient = createAsyncClient();
        }
    
        public static CloseableHttpAsyncClient getAsyncHttpClient(){
            if (null == asyncHttpClient){
                synchronized (LOCK){
                    if (null == asyncHttpClient){
                        asyncHttpClient = createAsyncClient();
                    }
                }
            }
            return asyncHttpClient;
        }
    
    
        private static CloseableHttpAsyncClient createAsyncClient(){
            RequestConfig requestConfig = RequestConfig.custom()
                    .setConnectTimeout(1000)
                    .setSocketTimeout(1000)
                    .setConnectionRequestTimeout(1000)
                    .build();
    
            //配置io线程
            IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
    //                .setIoThreadCount(10)
                    .setIoThreadCount(Runtime.getRuntime().availableProcessors())
                    .setSoKeepAlive(true)
                    .build();
            //设置连接池大小
            ConnectingIOReactor ioReactor = null;
            try {
                ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
            } catch (IOReactorException e) {
                e.printStackTrace();
            }
            Assert.notNull(ioReactor, "ioReactor init error");
            PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
            connManager.setMaxTotal(1000);
            connManager.setDefaultMaxPerRoute(100);
    
    
            final CloseableHttpAsyncClient client = HttpAsyncClients.custom()
                    .setConnectionManager(connManager)
                    .setDefaultRequestConfig(requestConfig)
                    .build();
            client.start();
            return client;
        }
    }

import com.sendinfo.datamanager.scheduler.exception.HttpException;
    import com.sendinfo.trace.core.util.LoggerFormatUtil;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.http.HttpResponse;
    import org.apache.http.NameValuePair;
    import org.apache.http.client.entity.UrlEncodedFormEntity;
    import org.apache.http.client.methods.HttpGet;
    import org.apache.http.client.methods.HttpPost;
    import org.apache.http.concurrent.FutureCallback;
    import org.apache.http.impl.client.HttpClients;
    import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
    import org.apache.http.impl.nio.client.HttpAsyncClients;
    import org.apache.http.message.BasicNameValuePair;
    import org.apache.http.nio.client.util.HttpAsyncClientUtils;
    import org.apache.http.util.EntityUtils;
    import org.springframework.util.CollectionUtils;
    
    import java.io.IOException;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Map;
    
    /**
     * @author
     */
    @Slf4j
    public class HttpAsyncUtils {
    
        public static void doGet(String url) throws HttpException {
            try {
                // 通过httpget方式来实现我们的get请求
                HttpGet httpGet = new HttpGet(url);
                CloseableHttpAsyncClient client = HttpAsyncClient.getAsyncHttpClient();
    //            CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();
    //            client.start();
                //异步请求
                client.execute(httpGet, new Back());
    //            client.close();
            } catch (Exception e) {
               
                throw new HttpException(e);
            }
        }
    
        public static void doPost(String url, Map<String, String> params) throws HttpException {
            try {
                // 通过HttpPost来发送post请求
                HttpPost httpPost = new HttpPost(url);
                /*
                 * post带参数开始
                 */
                // 第三步：构造list集合，往里面丢数据
                List<NameValuePair> list = new ArrayList<NameValuePair>();
                if (!CollectionUtils.isEmpty(params)){
                    for (Map.Entry<String, String> param : params.entrySet()){
                        BasicNameValuePair basicNameValuePair = new BasicNameValuePair(param.getKey(), param.getValue());
                        list.add(basicNameValuePair);
                    }
                }
                // 第二步：我们发现Entity是一个接口，所以只能找实现类，发现实现类又需要一个集合，集合的泛型是NameValuePair类型
                UrlEncodedFormEntity formEntity = new UrlEncodedFormEntity(list);
                // 第一步：通过setEntity 将我们的entity对象传递过去
                httpPost.setEntity(formEntity);
                CloseableHttpAsyncClient client = HttpAsyncClient.getAsyncHttpClient();
    //            CloseableHttpAsyncClient client = HttpAsyncClients.createDefault();
    //            client.start();
                //异步请求
                client.execute(httpPost, new Back());
    //            client.close();
            } catch (Exception e) {
                LoggerFormatUtil.error(e, log, "do post error");
                throw new HttpException(e);
            }
        }
    
       static class Back implements FutureCallback<HttpResponse> {
    
            private long start = System.currentTimeMillis();
    
            Back() {
            }
    
            @Override
            public void completed(HttpResponse httpResponse) {
                try {
                    System.out.println("cost is:" + (System.currentTimeMillis() - start) + ":" + EntityUtils.toString(httpResponse.getEntity()));
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            @Override
            public void failed(Exception e) {
                System.err.println(" cost is:" + (System.currentTimeMillis() - start) + ":" + e);
            }
            @Override
            public void cancelled() {
            }
        }
    }

<dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpclient</artifactId>
       <version>4.5.1</version>
    </dependency>
    
    <dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpcore</artifactId>
       <version>4.4.6</version>
    </dependency>
    
    <dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpmime</artifactId>
       <version>4.3.1</version>
    </dependency>
    
    <dependency>
       <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpasyncclient</artifactId>
       <version>4.1.3</version>
    </dependency>