基于netty的http client实现

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<groupId>syb</groupId>
	<artifactId>httpClientNetty</artifactId>
	<version>1.0</version>
	<packaging>jar</packaging>

	<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.1.1.RELEASE</version>
	</parent>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
		<java.version>1.8</java.version>
	</properties>

	<dependencies>

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter</artifactId>
		</dependency>

		<dependency>
			<groupId>io.netty</groupId>
			<artifactId>netty-all</artifactId>
		</dependency>

	</dependencies>

	<build>
		<finalName>httpClientNetty</finalName>
		<plugins>
			<plugin>
				<groupId>org.springframework.boot</groupId>
				<artifactId>spring-boot-maven-plugin</artifactId>
			</plugin>
		</plugins>
	</build>
</project>

启动类

装配netty，并发送一个http request

package syb.httpClientNetty;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultFullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpHeaderValues;
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpVersion;
import syb.httpClientNetty.httpclient.HttpChannelInitializer;

@Component
public class AppLifecycle implements SmartLifecycle {
    private Logger logger = LoggerFactory.getLogger(getClass());
    private boolean isRunning = false;
    
    @Autowired
    private HttpChannelInitializer cinit;

    @Override
    public void start() {
        logger.info("app start...");
        EventLoopGroup workGroup = null;
        String host = "192.168.5.3";
        int port = 9898;
        try {
            workGroup = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap()
                    .group(workGroup)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .channel(NioSocketChannel.class)
                    .handler(cinit);
            ChannelFuture f = bootstrap.connect(host, port).sync();
            Channel ch = f.channel();
            DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/test");
            request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE);
            request.headers().set(HttpHeaderNames.HOST, host);
            ch.writeAndFlush(request).sync();
            ch.closeFuture().sync();
        } catch (Exception e) {
            logger.error("", e);
        } finally {
            if (workGroup != null) {
                workGroup.shutdownGracefully();
            }
        }
        isRunning = true;
    }

    @Override
    public void stop() {
        try {
            
        } catch (Exception e) {
            logger.error("", e);
        }
        isRunning = false;
    }

    @Override
    public boolean isRunning() {
        return isRunning;
    }
}

通道初始化-加载通道处理器

package syb.httpClientNetty.httpclient;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;

/**
 * channel初始化器。负责装配channel的入站处理器。
 */
@Component
public class HttpChannelInitializer extends io.netty.channel.ChannelInitializer<NioSocketChannel> {
    @Autowired
    private SslHandlerFactory sslHandlerFactory;
    
    @Override
    protected void initChannel(NioSocketChannel ch) throws Exception {
        ch.pipeline().addLast(sslHandlerFactory.getSslHandler());
        ch.pipeline().addLast(new HttpClientCodec());
        ch.pipeline().addLast(new HttpObjectAggregator(1024 * 10 * 1024));
        ch.pipeline().addLast(new ClientHandler());
    }
}

创建SslHandler

package syb.httpClientNetty.httpclient;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.security.KeyStore;
import java.security.PublicKey;
import java.security.SecureRandom;
import java.security.cert.CertificateFactory;
import java.security.cert.X509Certificate;

import javax.net.ssl.KeyManager;
import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.TrustManager;

import org.springframework.boot.system.ApplicationHome;
import org.springframework.stereotype.Component;

import io.netty.handler.ssl.SslHandler;

@Component
public class SslHandlerFactory {
    private static String keyStorePassword = "baicells";

    public SslHandler getSslHandler() throws Exception {
        return createSslHandler();
    }

    private SslHandler createSslHandler() throws Exception {
        // 加载domain proxy证书
        KeyStore clientKeyStore = loadDpKeyStore();

        // 生成key mgr
        KeyManager[] kms = createKeyMgr(clientKeyStore);

        // 生成trust mgr
        TrustManager[] tms = createTrustMgr();

        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");

        // 初始化SSL Context
        sslContext.init(kms, tms, new SecureRandom());

        // SSL密码套件
        String[] ciperSuites = new String[] { "TLS_RSA_WITH_AES_128_GCM_SHA256",
                "TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256", "TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256" };

        SSLEngine engine = sslContext.createSSLEngine();
        engine.setUseClientMode(true);
        engine.setEnabledCipherSuites(ciperSuites);

        SslHandler sslHandler = new SslHandler(engine);
        return sslHandler;
    }

    private KeyStore loadDpKeyStore() throws Exception {
        KeyStore keyStore;
        InputStream in = null;
        ApplicationHome home = new ApplicationHome();
        try {
            String path = home.getDir().getAbsolutePath() + File.separator + "client.jks";
            in = new FileInputStream(path);
            keyStore = KeyStore.getInstance(KeyStore.getDefaultType());
            keyStore.load(in, keyStorePassword.toCharArray());
        } catch (FileNotFoundException e) {
            throw new FileNotFoundException(e.getMessage());
        } catch (Exception e) {
            throw new Exception(e);
        } finally {
            if (in != null) {
                in.close();
            }
        }

        return keyStore;
    }

    /**
     * 创建key mgr
     */
    private KeyManager[] createKeyMgr(KeyStore keyStore) throws Exception {
        KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509");
        kmf.init(keyStore, keyStorePassword.toCharArray());
        KeyManager[] kms = kmf.getKeyManagers();
        return kms;
    }

    /**
     * 创建trust mgr
     */
    private TrustManager[] createTrustMgr() throws Exception {
        MyX509TrustMgr trustMgr = null;

        try {
            PublicKey publicKey = getCaPublicKey();
            trustMgr = new MyX509TrustMgr(publicKey);
        } catch (FileNotFoundException e) {
            throw new FileNotFoundException();
        } catch (Exception e) {
            throw new Exception(e);
        }
        TrustManager[] tms = { trustMgr };
        return tms;
    }

    private PublicKey getCaPublicKey() throws Exception {
        PublicKey publicKey;
        InputStream in = null;
        ApplicationHome home = new ApplicationHome();
        try {
            String certPathForCa = home.getDir().getAbsolutePath() + File.separator + "trust.pem";
            in = new FileInputStream(certPathForCa);
            X509Certificate caCert = (X509Certificate) CertificateFactory.getInstance("X.509")
                    .generateCertificate(in);// 加载证书
            publicKey = caCert .getPublicKey();// 获取证书的public key
        } catch (Exception e) {
            throw new Exception(e);
        } finally {
            if (in != null) {
                in.close();
            }
        }
        return publicKey;
    }
}

自定义的X509TrustManager

package syb.httpClientNetty.httpclient;

import java.security.PublicKey;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;

import javax.net.ssl.X509TrustManager;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MyX509TrustMgr implements X509TrustManager {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private PublicKey caPublicKey = null;

    public MyX509TrustMgr() {
    }

    public MyX509TrustMgr(PublicKey ca) {
        this.caPublicKey = ca;
    }

    @Override
    public void checkClientTrusted(X509Certificate[] ax509certificate, String s) throws CertificateException {
        logger.info("checkClientTrusted");
    }

    @Override
    public void checkServerTrusted(X509Certificate[] certList, String authType) throws CertificateException {
        // 验证证书有效期
        for (X509Certificate cert : certList) {
            cert.checkValidity();
        }

        // 验证 CA
        checkCa(certList[0], certList[1]);
    }

    @Override
    public X509Certificate[] getAcceptedIssuers() {
        return null;
    }

    /**
     * 验证CA，如果验证不通过，抛出异常
     */
    private void checkCa(X509Certificate cert, X509Certificate caCert) throws CertificateException {
        logger.info("verify ca");
        
        if (caCert == null) {
            // 未读取到ca
            logger.error("no ca cert");
            throw new CertificateException("48:unknownCA");
        }
        
        if (!caCert.getPublicKey().equals(caPublicKey)) {
            // 读取到的ca与信任的ca不匹配
            logger.error("ca cert not equal trusted ca cert");
            throw new CertificateException("48:unknownCA");
        }
        
        String issuerDN = cert.getIssuerDN().toString();
        String caDN = caCert.getSubjectDN().toString();
        if (!issuerDN.equals(caDN)) {
            throw new CertificateException("48:unknownCA");
        } else {
            try {
                cert.verify(caCert.getPublicKey());
            } catch (Exception e) {
                throw new CertificateException("51:decryptError");
            }
        }

        logger.info("verify ca, ok");

    }
}

http response处理器

只是简单的打印http response信息。

package syb.httpClientNetty.httpclient;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;

public class ClientHandler extends ChannelInboundHandlerAdapter {
    private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        logger.info("receive...");
        FullHttpResponse response = (FullHttpResponse) msg;
        ByteBuf content = response.content();
        HttpHeaders headers = response.headers();
        
        logger.info("headers:{}", headers.toString());
        byte[] data = new byte[content.readableBytes()];
        content.readBytes(data);
        logger.info("content:{}", new String(data));
    }

}