如下为一个使用了线程池，SSL，出现指定异常后充实，超时配置的demo

SSL配置

packagecom.demo.client;importjava.io.FileInputStream;importjava.io.InputStream;importjava.security.KeyStore;importjava.security.PublicKey;importjava.security.cert.CertificateFactory;importjava.security.cert.X509Certificate;importjava.util.ArrayList;importjava.util.LinkedList;importjava.util.List;importjava.util.function.Consumer;importjavax.net.ssl.KeyManagerFactory;importjavax.net.ssl.SNIMatcher;importjavax.net.ssl.SNIServerName;importjavax.net.ssl.SSLEngine;importjavax.net.ssl.SSLParameters;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importio.netty.handler.ssl.SslContext;importio.netty.handler.ssl.SslContextBuilder;importreactor.netty.tcp.SslProvider.SslContextSpec;public class SslConsumer implements Consumer{private Logger logger = LoggerFactory.getLogger(SslConsumer.class);privateString keyStorePath;private String password = "123456";private boolean trustServer = true;//测试，默认为true

private String serverCaPath = "";publicSslConsumer(String keyStorePath) {this.keyStorePath =keyStorePath;

}

@Overridepublic voidaccept(SslContextSpec t) {try{

t.sslContext(createSslContext(loadKeyStore(keyStorePath))).handlerConfigurator(handler->{

SSLEngine engine=handler.engine();

List matchers = new LinkedList();

SNIMatcher matcher= new SNIMatcher(0) {

@Overridepublic booleanmatches(SNIServerName serverName) {//返回true，不验证主机名

return true;

}

};

matchers.add(matcher);

SSLParameters params= newSSLParameters();

params.setSNIMatchers(matchers);

engine.setSSLParameters(params);

});

}catch(Exception e) {

}

}private SslContext createSslContext(KeyStore keyStore) throwsException {

SslContextBuilder builder=SslContextBuilder.forClient();

KeyManagerFactory keyMgrFactory= KeyManagerFactory.getInstance("SunX509");

keyMgrFactory.init(keyStore, password.toCharArray());

builder.keyManager(keyMgrFactory);

SSLX509TrustMgr trustMgr= null;if(trustServer) {

trustMgr= newSSLX509TrustMgr();

}else{

trustMgr= newSSLX509TrustMgr(getSeverPublicKey(serverCaPath));

}

builder.trustManager(trustMgr);

List ciper = new ArrayList();

ciper.add("TLS_RSA_WITH_AES_128_GCM_SHA256");

ciper.add("TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256");

ciper.add("TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256");

builder.ciphers(ciper);returnbuilder.build();

}privateKeyStore loadKeyStore(String keyStorePath) {

KeyStore keyStore= null;

InputStream in= null;try{

in= newFileInputStream(keyStorePath);

keyStore=KeyStore.getInstance(KeyStore.getDefaultType());

keyStore.load(in, password.toCharArray());

}catch(Exception e) {

logger.error("", e);

}returnkeyStore;

}privatePublicKey getSeverPublicKey(String serverCaPath) {

PublicKey key= null;

InputStream in= null;try{

in= newFileInputStream(serverCaPath);

X509Certificate cert= (X509Certificate) CertificateFactory.getInstance("X.509").generateCertificate(in);

key=cert.getPublicKey();

}catch(Exception e) {

logger.error("", e);

}returnkey;

}

}

验证服务器整数实现类

目前没有做任何处理，有需要验证服务器证书时，可以在对应的重写方法中进行处理

packagecom.demo.client;importjava.security.PublicKey;importjava.security.cert.CertificateException;importjava.security.cert.X509Certificate;importjavax.net.ssl.X509TrustManager;public class SSLX509TrustMgr implementsX509TrustManager {privatePublicKey serverPublicKey;publicSSLX509TrustMgr() {

}publicSSLX509TrustMgr(PublicKey serverPublicKey) {this.serverPublicKey =serverPublicKey;

}

@Overridepublic void checkClientTrusted(X509Certificate[] chain, String authType) throwsCertificateException {//TODO Auto-generated method stub

}

@Overridepublic void checkServerTrusted(X509Certificate[] chain, String authType) throwsCertificateException {//TODO Auto-generated method stub

}

@OverridepublicX509Certificate[] getAcceptedIssuers() {//TODO Auto-generated method stub

return null;

}

}

WebClient工具类

packagecom.demo.utils;importjava.net.ConnectException;importjava.net.NoRouteToHostException;importjava.time.Duration;importjava.util.Map;importjava.util.concurrent.ConcurrentHashMap;importorg.slf4j.Logger;importorg.slf4j.LoggerFactory;importorg.springframework.http.MediaType;importorg.springframework.http.client.reactive.ReactorClientHttpConnector;importorg.springframework.web.reactive.function.client.WebClient;importcom.demo.entity.SslConsumer;importio.netty.channel.ChannelOption;importio.netty.channel.ConnectTimeoutException;importio.netty.handler.timeout.ReadTimeoutHandler;importio.netty.handler.timeout.WriteTimeoutHandler;importreactor.core.publisher.Mono;importreactor.netty.http.client.HttpClient;importreactor.netty.resources.ConnectionProvider;importreactor.netty.resources.LoopResources;importreactor.netty.tcp.TcpClient;importreactor.retry.Backoff;importreactor.retry.Retry;public classWebClientUtil {private static Logger logger = LoggerFactory.getLogger(WebClientUtil.class);private static Map webClientMap = new ConcurrentHashMap();public staticWebClient createWebClient(String keyStorePath) {

ConnectionProvider provider= ConnectionProvider.builder("wc-").maxConnections(30)

.maxIdleTime(Duration.ofSeconds(30)).maxLifeTime(Duration.ofSeconds(30))

.pendingAcquireTimeout(Duration.ofSeconds(30)).build();

LoopResources loop= LoopResources.create("loop-", 20, 20, true);

TcpClient tcpClient= TcpClient.create(provider).secure(newSslConsumer(keyStorePath)).runOn(loop)

.option(ChannelOption.CONNECT_TIMEOUT_MILLIS,2000).doOnConnected(conn ->conn

.addHandlerLast(new ReadTimeoutHandler(30)).addHandlerLast(new WriteTimeoutHandler(20)));return WebClient.builder().clientConnector(newReactorClientHttpConnector(HttpClient.from(tcpClient))).build();//如下方式可能会一直循环生成loop线程，直到程序僵死//HttpClient httpClient = HttpClient.create(provider).secure(new SslConsumer(keyStorePath))//.tcpConfiguration(client -> client.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)//.doOnConnected(con -> con.addHandlerLast(new ReadTimeoutHandler(20))//.addHandlerLast(new WriteTimeoutHandler(10)))//.runOn(LoopResources.create("loop-", 20, 20, true)));//

//return WebClient.builder().clientConnector(new ReactorClientHttpConnector(httpClient)).build();

}public static MonodoPost(String url, String data, String keyStorePath) {

WebClient webClient=webClientMap.get(url);if (webClient == null) {

webClient=createWebClient(keyStorePath);

WebClient putIfAbsent=webClientMap.putIfAbsent(url, webClient);if (putIfAbsent != null) {

webClient=putIfAbsent;

}

}

Retry> retry = Retry.anyOf(ConnectTimeoutException.class, NoRouteToHostException.class).retryMax(3)

.backoff(Backoff.fixed(Duration.ofMillis(100)));

Mono mono =webClient.post().uri(url).contentType(MediaType.APPLICATION_JSON).bodyValue(data).retrieve()

.bodyToMono(String.class).timeout(Duration.ofSeconds(30)).doOnError(ConnectException.class, e ->{

logger.error("", e);

}).doOnError(NoRouteToHostException.class, e ->{

logger.error("", e);

}).retryWhen(retry);returnmono;

}

}

1、demo使用的springboot版本为2.3.3

2、HttpClient的create方法与secure方法不能分开，否则secure方法可能不生效

3、 ConnectionProvider中有很多方法已经过时，比如fixed，elastic，不建议使用，在过时的方法上，源码中也给出了对应替换使用的方法，比如fixed可以使用create方法替代，elastic方法可以用builder方法替代，但create方法直接创建的就为ConnectionProvider，重载的方法可以设置maxConnections，但不能设置连接最大空闲时间，连接最大生命周期等，builder方法可以对线程更精细的管理，故本例使用的builder方法和build来创建ConnectionProvider，如果使用默认的provider，线程数默认500，且maxIdleTime和maxLifeTime为-1

4、LoopResources为处理响应消息的线程数，2.3.3版本最小值为4

5、ReactorClientHttpConnector存在一个重构方法

在该方法中，factory可以设置ConnectionProvider和LoopResources，mapper可以设置HttpClient，但在使用时可能是使用方式不正确，一直都是要么线程池配置上了，但SSL不可用，或者是SSL可用，但线程池不可用，后在ReactorClientHttpConnector两个参数的构造方法中看到initHttpClient方法，该方法中，使用了runOn配置LoopResources，故本例中也是用了runOn来配置。

测试：

@Overridepublic void run(String... args) throwsException {int count = 6;for (int i = 0; i < count; i++) {

logger.info("start: {}", i);

JsonObject obj= newJsonObject();

obj.addProperty("name", "name"+i);

obj.addProperty("age", i);

Mono mono =WebClientUtil.doPost(url, obj.toString(), keyStorePath);

mono.subscribe(newMainConsumer(i));

logger.info("end: {}", i);

}

}class MainConsumer implements Consumer{private inti;public MainConsumer(inti) {this.i =i;

}

@Overridepublic voidaccept(String t) {

logger.info("receive: {}, loop: {}", t, i);

}

}

View Code

结果：

[INFO] main 14:44:30 WebclientdemoApplication:61(logStarted) Started WebclientdemoApplication in 3.453 seconds (JVM running for 4.137)

[INFO] main 14:44:30 MainBusiStart:29(run) start: 0

[INFO] main 14:44:31 MainBusiStart:35(run) end: 0

[INFO] main 14:44:31 MainBusiStart:29(run) start: 1

[INFO] main 14:44:31 MainBusiStart:35(run) end: 1

[INFO] main 14:44:31 MainBusiStart:29(run) start: 2

[INFO] main 14:44:31 MainBusiStart:35(run) end: 2

[INFO] main 14:44:31 MainBusiStart:29(run) start: 3

[INFO] main 14:44:31 MainBusiStart:35(run) end: 3

[INFO] main 14:44:31 MainBusiStart:29(run) start: 4

[INFO] main 14:44:31 MainBusiStart:35(run) end: 4

[INFO] main 14:44:31 MainBusiStart:29(run) start: 5

[INFO] main 14:44:31 MainBusiStart:35(run) end: 5

[INFO] loop--nio-4 14:44:31 MainBusiStart:48(accept) receive: {"name":"name3","age":3.0}, loop: 3

[INFO] loop--nio-5 14:44:31 MainBusiStart:48(accept) receive: {"name":"name4","age":4.0}, loop: 4

[INFO] loop--nio-1 14:44:31 MainBusiStart:48(accept) receive: {"name":"name0","age":0.0}, loop: 0

[INFO] loop--nio-2 14:44:31 MainBusiStart:48(accept) receive: {"name":"name1","age":1.0}, loop: 1

[INFO] loop--nio-3 14:44:31 MainBusiStart:48(accept) receive: {"name":"name2","age":2.0}, loop: 2

[INFO] loop--nio-6 14:44:31 MainBusiStart:48(accept) receive: {"name":"name5","age":5.0}, loop: 5

View Code

6、本例中，使用了一个reactor-extra进行重试处理

io.projectreactor.addons

reactor-extra

但这种重试的方法虽然比较简单好用，但已经被标记为过时

应该使用传递一个reactor.util.retry.Retry类型参数的方法

7、如果使用springclloud，则需要考虑springboot和springcloud的兼容问题，以及当前程序的springboot和springcloud版本与其他需要远程调用的版本的兼容