一款高性能的穿透网闸架构

1 前言

2 设计

2.1 技术选型

2.1.1 使用kafka作为不同网络之间通讯工具

2.1.2 重构网关

2.1.3 Nacos注册中心

2.1.4 zookpeer使用

2.1.4 架构设计

2.1.5 项目架构图

3 服务说明

3.1 igferry-deliver服务

3.2 igferry-gateway网关服务

MASTER_POOL.submit(() -> {
	try {
		Request request = JacksonUtil.fromJson(value, Request.class);
		SpringBeanUtils.getBean(KafkaMessageDynamicLoad.class).loadMQMsg(request).block();
 } catch (Exception e) {
		e.printStackTrace();
 log.error("handle msg error topic: {}, value: {} ,errormsg: {}", new Object[]{consumerTopic, value, e.getMessage()});
 }
});

public Mono<Void> loadMQMsg(Request<?> request){
 String appName = parseAppName(request);
 MQInvokerChain mqInvokerChain = new MQInvokerChain(serverConfigProperties,gatewayKafkaConfig,grayLoadBalancer
 ,gatewayKafkaConnect,appName);
 mqInvokerChain.addPlugin(new MsgAuthMQInvoker(serverConfigProperties,gatewayKafkaConfig,grayLoadBalancer,gatewayKafkaConnect));
 mqInvokerChain.addPlugin(new MsgDynamicRouteMQInvoker(serverConfigProperties,gatewayKafkaConfig,grayLoadBalancer,gatewayKafkaConnect));
 return mqInvokerChain.execute(request,mqInvokerChain);
}

public class MQInvokerChain extends AbstractMQInvokerImpl {
 /**
 * 服务id
 */
 private String appName;
 /**
 * 当前执行的链路插件
 */
 private int pos;
 /**
 * 存放服务链路
 */
 private List<MQInvoker> mqInvokers;
 public MQInvokerChain(ServerConfigProperties serverConfigProperties, GatewayKafkaConfig gatewayKafkaConfig,
 GrayLoadBalancer grayLoadBalancer,
 GatewayKafkaConnect gatewayKafkaConnect,String appName) {
 super(serverConfigProperties, gatewayKafkaConfig,grayLoadBalancer, gatewayKafkaConnect);
 this.appName = appName;
 }
 /**
 * 将启用的插件添加到链
 *
 * @param mqInvoker
 */
 public void addPlugin(MQInvoker mqInvoker) {
 if (mqInvokers == null) {
 mqInvokers = new ArrayList<>();
 }
 mqInvokers.add(mqInvoker);
 // 排序
 mqInvokers.sort(Comparator.comparing(MQInvoker::order));
 }
 @Override
 public Integer order() {
 return 0;
 }
 @Override
 public String name() {
 return null;
 }
 /**
 * 执行调用链
 */
 @Override
 public Mono<Void> execute(Request<?> request, MQInvokerChain mqInvokerChain) {
 if (pos == mqInvokers.size()) {
 return Mono.empty();
 }
 return mqInvokerChain.mqInvokers.get(pos++).execute(request, mqInvokerChain);
 }
 public String getAppName() {
 return appName;
 }
}

public class MsgDynamicRouteMQInvoker extends AbstractMQInvokerImpl {
 private static WebClient webClient;
 public MsgDynamicRouteMQInvoker(ServerConfigProperties serverConfigProperties, GatewayKafkaConfig gatewayKafkaConfig,
 GrayLoadBalancer grayLoadBalancer,
 GatewayKafkaConnect gatewayKafkaConnect) {
 super(serverConfigProperties, gatewayKafkaConfig, grayLoadBalancer, gatewayKafkaConnect);
 }
 static {
 HttpClient httpClient = HttpClient.create()
 .tcpConfiguration(client ->
 client.doOnConnected(conn ->
 conn.addHandlerLast(new ReadTimeoutHandler(180))
 .addHandlerLast(new WriteTimeoutHandler(180)))
 .option(ChannelOption.TCP_NODELAY, true)
 );
 webClient = WebClient.builder().clientConnector(new ReactorClientHttpConnector(httpClient))
 .build();
 }
 @Override
 public Integer order() {
 return MQInvokerEnum.DYNAMIC_FERRY_ROUTE.getOrder();
 }
 @Override
 public String name() {
 return MQInvokerEnum.DYNAMIC_FERRY_ROUTE.name();
 }
 @Override
 public Mono<Void> execute(Request<?> request, MQInvokerChain pluginChain) {
 log.info("mq消息路由转发");
 String appName = pluginChain.getAppName();
 ServiceInstance serviceInstance = chooseInstance(appName);
 String url = buildUrl(serviceInstance, request.getApiUrl());
 return forward(request, url);
 }
 private Mono<Void> forward(Request<?> request, String url) {
 HttpMethod httpMethod = HttpMethod.valueOf(request.getHttpMethod());
 WebClient.RequestBodySpec requestBodySpec = webClient.method(httpMethod).uri(url);
 WebClient.RequestHeadersSpec<?> reqHeadersSpec;
 if (requireHttpBody(httpMethod)) {
 reqHeadersSpec = requestBodySpec.contentType(MediaType.APPLICATION_JSON).body(BodyInserters.fromValue(request.getBody()));
 } else {
 reqHeadersSpec = requestBodySpec;
 }
 // 消息回调
 return getResultMsg(request, reqHeadersSpec).flatMap(response -> {
 gatewayKafkaConnect.sendCoustomKafka(response.getTopic(), request.getKey(), JacksonUtil.toJson(response));
 return Mono.empty();
 });
 }
 // nio->callback->nio
 private Mono<Response> getResultMsg(Request<?> request, WebClient.RequestHeadersSpec<?> reqHeadersSpec) {
 Mono<Response> responseMono = reqHeadersSpec.exchangeToMono(clientResponse -> {
 return clientResponse.bodyToMono(String.class).flatMap(responseBody -> {
 Response response = JacksonUtil.fromJson(responseBody, Response.class);
 response.setTopic(request.getResponseTopic());
 return Mono.just(response);
 });
 });
 if (request.isAsync()) {
 return responseMono.timeout(Duration.ofMillis(serverConfigProperties.getTimeOutMillis()))
 .onErrorResume(ex -> {
 return Mono.defer(() -> {
 Response response = new Response();
 response.setTopic(request.getResponseTopic());
 if (ex instanceof TimeoutException) {
 response.setErrCode("5001");
 response.setErrMsg("network timeout");
 } else {
 response.setErrCode("5000");
 response.setErrMsg("system error");
 }
 return Mono.just(response);
 }).then(Mono.just(new Response(request.getResponseTopic(), "5000", "system error")));
 });
 }
 return responseMono;
 }
 private String buildUrl(ServiceInstance serviceInstance, String apiUrl) {
 String path = apiUrl.replaceFirst("/" + serviceInstance.getServiceId(), "");
 String url = "http://" + serviceInstance.getHost() + ":" + serviceInstance.getPort() + path;
 return url;
 }
 /**
 * 查看http请求是否需要参数body
 * @param method
 * @return
 */
 private boolean requireHttpBody(HttpMethod method) {
 if (method.equals(HttpMethod.POST) || method.equals(HttpMethod.PUT) || method.equals(HttpMethod.PATCH)) {
 return true;
 }
 return false;
 }
 /**
 * 根据路由规则配置和负载均衡算法选择服务实例
 *
 * @param appName
 * @return
 */
 private ServiceInstance chooseInstance(String appName) {
 return grayLoadBalancer.choose(appName);
 }
}

3.3 igferry-server服务

4 本地调试

4.1 本地启动nacos服务。

git clone https://github.com/alibaba/nacos.git
cd nacos/
mvn -Prelease-nacos -Dmaven.test.skip=true clean install -U 
ls -al distribution/target/
// change the $version to your actual path
cd distribution/target/nacos-server-$version/nacos/bin
//启动nacos
startup.cmd -m standalone

4.2 启动zookeeper和kafka

4.3 启动两个igferry-gateway

4.4 启动两个igferry-server

4.5 启动igferry-deliver

4.6 用postman调试链路是否正常

4.7 性能压测

5.总结

参考资料