soyo的跨机房部署、同城双活、异地多活、9个9高可用博客yuque笔记同步site

public String getCity(User user) throws Exception{  
    return Optional.ofNullable(user)  
                   .map(u-> u.getAddress())  
                   .map(a->a.getCity())  
                   .orElseThrow(()->new Exception("取指错误"));  
} 

@RestController
@ApiService(serviceId = "backend-http-server", protocol = ApiProtocol.HTTP, patternPath = "/http-server/**")
@Slf4j
public class HttpController {

	@Autowired
	private ApiProperties apiProperties;

	@ApiInvoker(path = "/http-server/ping")
	@GetMapping("/http-server/ping")
	public String ping() {
		log.info("{}", apiProperties);
		return "pong1";
	}
}

/**
 * 默认路由逻辑：
 * 根据 whenComplete 	判断执行回调的线程是否阻塞执行；
 * whenComplete 		当异步操作完成时（无论成功还是失败），会立即执行回调函数；
 * whenCompleteAsync 	当异步操作完成时，会创建一个新的异步任务来执行回调函数。
 */
private CompletableFuture<Response> route(GatewayContext gatewayContext, Optional<Rule.HystrixConfig> hystrixConfig) {
    // 异步请求发送
    Request request = gatewayContext.getRequest().build();
    CompletableFuture<Response> future = AsyncHttpHelper.getInstance().executeRequest(request);
    boolean whenComplete = ConfigLoader.getConfig().isWhenComplete();

    // 异步/非异步模型
    if (whenComplete) {
        future.whenComplete(((response, throwable) -> {
            complete(request, response, throwable, gatewayContext);
        }));
    } else {
        future.whenCompleteAsync(((response, throwable) -> {
            complete(request, response, throwable, gatewayContext);
        }));
    }
    return future;
}

private void handleResponse(Request request, Response response, Throwable throwable, GatewayContext gatewayContext) {
	String url = request.getUrl();
	try {
		if (Objects.nonNull(throwable)) {
			if (throwable instanceof TimeoutException) {
				log.warn("complete timeout {}", url);
				gatewayContext.setThrowable(throwable);
				gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.REQUEST_TIMEOUT));
			} else if (throwable instanceof IOException) {
				gatewayContext.setThrowable(new ConnectException(throwable, gatewayContext.getUniqueId(), url, ResponseCode.HTTP_RESPONSE_ERROR));
				gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.HTTP_RESPONSE_ERROR));
			}
		} else {
			gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(response));
		}
	} catch (Exception e) {
		gatewayContext.setThrowable(new ResponseException(ResponseCode.INTERNAL_ERROR));
		gatewayContext.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.INTERNAL_ERROR));
		log.error("complete process failed", e);
	} finally {
		gatewayContext.setContextStatus(ContextStatus.Written);
		ResponseHelper.writeResponse(gatewayContext);
	}
}

/**
 * 写回响应
 */
public static void writeResponse(IContext context) {
    context.releaseRequest();

    if (context.judgeContextStatus(ContextStatus.Written)) {
        FullHttpResponse response = getHttpResponse(context, (GatewayResponse) context.getResponse());
        // 如果不是保持连接的情况，响应后关闭通道
        if (!context.isKeepAlive()) {
            context.getNettyContext().writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
        } else {
            response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
            context.getNettyContext().writeAndFlush(response);
        }
        context.setContextStatus(ContextStatus.Completed);
    } else if (context.judgeContextStatus(ContextStatus.Completed)) {
        context.invokeCompletedCallBacks();
    }
}

private void complete(Request request, Response response, Throwable throwable, GatewayContext gatewayContext) {
	// 请求已经处理完毕 释放请求资源
	gatewayContext.releaseRequest();
	// 获取上下文请求配置规则
	Rule rule = gatewayContext.getRules();
	// 获取重试次数
	int currentRetryTimes = gatewayContext.getCurrentRetryTimes();
	int confRetryTimes = rule.getRetryConfig().getTimes();
	// 异常重试
	if ((throwable instanceof TimeoutException
			|| throwable instanceof IOException)
			&& currentRetryTimes <= confRetryTimes) {
		doRetry(gatewayContext, currentRetryTimes);
	}
	// 处理响应
	handleResponse(request, response, throwable, gatewayContext);
}

/**
 * 重试策略
 */
private void doRetry(GatewayContext gatewayContext, int retryTimes) {
	gatewayContext.setCurrentRetryTimes(retryTimes + 1);
	log.info("当前请求重试次数为{}", gatewayContext.getCurrentRetryTimes());
	try {
		// 重新执行过滤器逻辑
		doFilter(gatewayContext);
	} catch (Exception e) {
		log.warn("重试请求失败, requestId={}", gatewayContext.getUniqueId(), e);
		throw new RuntimeException(e);
	}
}

private static Optional<Rule.HystrixConfig> getHystrixConfig(GatewayContext gatewayContext) {
	Rule rule = gatewayContext.getRules();
	Optional<Rule.HystrixConfig> hystrixConfig = rule.getHystrixConfigs().stream()
			.filter(c -> StringUtils.equals(c.getPath(), gatewayContext.getRequest().getPath()))
			.findFirst();
	return hystrixConfig;
}

private void routeWithHystrix(GatewayContext gatewayContext, Optional<Rule.HystrixConfig> hystrixConfig) {
    String key = gatewayContext.getUniqueId() + "." + gatewayContext.getRequest().getPath();
    RouterHystrixCommand proxyCommand = null;

    if (commandMap.containsKey(key)) {
        proxyCommand = commandMap.get(key);
        if (!hystrixConfig.get().equals(commandMap.get(key))) {
            log.info("previous HystrixCommand instance hashCode: {}", proxyCommand.hashCode());
            proxyCommand.updateHystrixCommandProperties(proxyCommand.getCommandKey().name());
            proxyCommand = new RouterHystrixCommand(gatewayContext, hystrixConfig);

            log.info("after HystrixCommand instance hashCode: {}", proxyCommand.hashCode());
            commandMap.put(key, proxyCommand);
        }
    } else {
        proxyCommand = new RouterHystrixCommand(gatewayContext, hystrixConfig);
        commandMap.put(key, proxyCommand);
    }
    proxyCommand.execute();
}

/**
 * Hystrix命令集合
 */
private class RouterHystrixCommand extends HystrixCommand<Object> {
    private GatewayContext context;
    private Optional<Rule.HystrixConfig> config;
    public RouterHystrixCommand(GatewayContext context, Optional<Rule.HystrixConfig> config) {
        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(context.getUniqueId()))
              .andCommandKey(HystrixCommandKey.Factory.asKey(context.getRequest().getPath()))
              .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter()
                                               // 核心线程数
                                               .withCoreSize(config.get().getCoreThreadSize()))
              .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                                            // 线程隔离类型
                                            .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.THREAD)
                                            // 命令执行超时
                                            .withExecutionTimeoutInMilliseconds(config.get().getTimeoutInMilliseconds())
                                            // 超时中断
                                            .withExecutionIsolationThreadInterruptOnTimeout(true)
                                            .withExecutionTimeoutEnabled(true)));
        this.config = config;
        this.context = context;
    }
    @Override
    protected Object run() throws Exception {
        // 实际路由操作
        route(context, config).get();
        return null;
    }
    /**
	 * 熔断降级操作
	 */
    @Override
    protected Object getFallback() {
        // 是否是超时引发的熔断
        if (isFailedExecution() || getExecutionException() instanceof HystrixTimeoutException) {
            // 针对超时的异常处理
            context.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.GATEWAY_FALLBACK_TIMEOUT));
        } else {
            // 其它类型异常熔断处理
            context.setResponse(GatewayResponse.buildGatewayResponse(ResponseCode.GATEWAY_FALLBACK_ERROR, config.get().getFallbackResponse()));
        }
        context.setContextStatus(ContextStatus.Written);
        return null;
    }
    /**
	 * 动态更新 CommandProperties 配置
	 * 1.因为 Hystrix 内部使用了缓存，如果仅仅修改 HystrixCommand.Setter 是没有用的；
	 * 2.利用反射获取 HystrixPropertiesFactory 的 commandProperties 字段，并更新
	 */
    protected void updateHystrixCommandProperties(String commandKey) {
        try {
            Field field = HystrixPropertiesFactory.class.getDeclaredField("commandProperties");
            field.setAccessible(true);
            ConcurrentHashMap<String, HystrixCommandProperties> commandProperties = (ConcurrentHashMap<String, HystrixCommandProperties>) field.get(null);
            log.info("before update HystrixCommandProperties: {}", commandProperties.get(commandKey));
            commandProperties.remove(commandKey);
        } catch (NoSuchFieldException | IllegalAccessException e) {
            log.error("Remove cache in HystrixCommandFactory failed, commandKey: {}", commandKey, e);
        }
    }
}

/**
 * SPI加载本地过滤器实现类对象
 * 过滤器存储映射 过滤器id - 过滤器对象
 */
public GatewayFilterChainFactory() {
	//加载所有过滤器
	ServiceLoader<Filter> serviceLoader = ServiceLoader.load(Filter.class);
	serviceLoader.stream().forEach(filterProvider -> {
		Filter filter = filterProvider.get();
		FilterAspect annotation = filter.getClass().getAnnotation(FilterAspect.class);
		log.info("load filter success:{},{},{},{}", filter.getClass(), annotation.id(), annotation.name(), annotation.order());
		//添加到过滤集合
		String filterId = annotation.id();
		if (StringUtils.isEmpty(filterId)) {
			filterId = filter.getClass().getName();
		}
		processorFilterIdMap.put(filterId, filter);
		processFilterIdName.put(filterId, annotation.name());
	});
}

/**
 * 过滤器链工厂
 */
private FilterChainFactory chainFactory = GatewayFilterChainFactory.getInstance();

// 创建并填充 GatewayContext 以保存有关传入请求的信息。
GatewayContext gatewayContext = RequestHelper.doContext(request, ctx);

// 组装过滤器并执行过滤操作
chainFactory.buildFilterChain(gatewayContext).doFilter(gatewayContext);

/**
 * 过滤器链缓存（服务ID ——> 过滤器链）
 * ruleId —— GatewayFilterChain
 */
private Cache<String, GatewayFilterChain> chainCache = Caffeine.newBuilder().recordStats().expireAfterWrite(10, TimeUnit.SECONDS).build();

/**
 * 构建过滤器链条
 */
@Override
public GatewayFilterChain buildFilterChain(GatewayContext ctx) throws Exception {
    // 获取规则ID
    String ruleId = ctx.getRules().getId();

    // 从缓存中获取过滤器链
    GatewayFilterChain chain = chainCache.getIfPresent(ruleId);

    // 如果缓存中没有过滤器链，那么构建一个新的过滤器链
    if (chain == null) {
        chain = doBuildFilterChain(ctx.getRules());
        // 将新构建的过滤器链添加到缓存中
        chainCache.put(ruleId, chain);
    }

    // 返回过滤器链
    return chain;
}

// 组装过滤器并执行过滤操作
chainFactory.buildFilterChain(gatewayContext).doFilter(gatewayContext);

/**
 * 执行过滤器链
 */
public GatewayContext doFilter(GatewayContext ctx) {
	if (filters.isEmpty()) {
		return ctx;
	}
	try {
		for (Filter filter : filters) {
			filter.doFilter(ctx);
		}
	} catch (Exception e) {
		log.error("执行过滤器发生异常: {}", e.getMessage());
		throw new RuntimeException(e);
	}
	return ctx;
}