Midnight

Soul网关中的Apache Dubbo插件执行原理（一）

29 Jan 2021   |   Soul  |  

在之前的文章中，我们体验过了Apache Dubbo插件执行流程，本篇文章是通过跟踪源码的方式来理解其中的执行原理。

在Soul网关中，Apache Dubbo插件负责将http协议转换成dubbo协议，设计到的插件有：BodyParamPlugin，ApacheDubboPlugin和DubboResponsePlugin。

  • BodyParamPlugin：负责将请求的json放到exchange属性中；
  • ApacheDubboPlugin：使用Apache Dubbo进行请求的泛化调用并返回响应结果；
  • DubboResponsePlugin：包装响应结果。

BodyParamPlugin插件

该插件在执行链路中是先执行的，负责处理请求类型，比如带有参数的application/json，不带参数的查询请求。

//org.dromara.soul.plugin.apache.dubbo.param.BodyParamPlugin#execute   
public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        //省略了其他代码
        if (Objects.nonNull(soulContext) && RpcTypeEnum.DUBBO.getName().equals(soulContext.getRpcType())) {
           //省略了其他代码
            //处理application/json
            if (MediaType.APPLICATION_JSON.isCompatibleWith(mediaType)) {
                return body(exchange, serverRequest, chain);
            }
            //处理application/x-www-form-urlencoded
            if (MediaType.APPLICATION_FORM_URLENCODED.isCompatibleWith(mediaType)) {
                return formData(exchange, serverRequest, chain);
            }
            //处理查询请求
            return query(exchange, serverRequest, chain);
        }
        return chain.execute(exchange);
    }

ApacheDubboPlugin插件

完成ApacheDubbo插件的核心处理逻辑：检查元数据 -->检查参数类型-->泛化调用。

//org.dromara.soul.plugin.apache.dubbo.ApacheDubboPlugin#doExecute 
@Override
    protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
        String body = exchange.getAttribute(Constants.DUBBO_PARAMS);
        SoulContext soulContext = exchange.getAttribute(Constants.CONTEXT);
        assert soulContext != null;
        MetaData metaData = exchange.getAttribute(Constants.META_DATA);
        //检查元数据
        if (!checkMetaData(metaData)) {
			//省略了其他代码
        }
        //检查参数类型
        if (StringUtils.isNoneBlank(metaData.getParameterTypes()) && StringUtils.isBlank(body)) {
            exchange.getResponse().setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
            Object error = SoulResultWrap.error(SoulResultEnum.DUBBO_HAVE_BODY_PARAM.getCode(), SoulResultEnum.DUBBO_HAVE_BODY_PARAM.getMsg(), null);
            return WebFluxResultUtils.result(exchange, error);
        }
        //泛化调用
        final Mono<Object> result = dubboProxyService.genericInvoker(body, metaData, exchange);
        //执行下一个插件
        return result.then(chain.execute(exchange));
    }

genericInvoker()方法中参数分别是请求参数body，有关服务信息的metaData，包含web信息的exchange。这里面的主要操作是：

  • 根据请求路径从缓存获取服务配置信息；
  • 获取代理对象；
  • 请求参数转化为dubbo泛化参数；
  • apache dubbo真正的泛化调用；
  • 返回结果。
public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
		//省略了其他代码
   	   //根据请求路径从缓存获取服务配置信息
        ReferenceConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());

   		 //获取代理对象
        GenericService genericService = reference.get();
    	//请求参数转化为dubbo泛化参数
        Pair<String[], Object[]> pair;
        if (ParamCheckUtils.dubboBodyIsEmpty(body)) {
            pair = new ImmutablePair<>(new String[]{}, new Object[]{});
        } else {
            pair = dubboParamResolveService.buildParameter(body, metaData.getParameterTypes());
        }
     
    	//apache dubbo真正的泛化调用，注意这里是异步的泛化调用
        CompletableFuture<Object> future = genericService.$invokeAsync(metaData.getMethodName(), pair.getLeft(), pair.getRight());
    
    //返回结果
        return Mono.fromFuture(future.thenApply(ret -> {
            if (Objects.isNull(ret)) {
                ret = Constants.DUBBO_RPC_RESULT_EMPTY;
            }
            exchange.getAttributes().put(Constants.DUBBO_RPC_RESULT, ret);
            exchange.getAttributes().put(Constants.CLIENT_RESPONSE_RESULT_TYPE, ResultEnum.SUCCESS.getName());
            return ret;
        }))//省略了其他代码
    }

代码中reference的信息如下，这里明确可以看出是dubbo协议，有接口名称，超时时间，重试次数，负载均衡等信息。

<dubbo:reference protocol="dubbo" prefix="dubbo.reference.org.dromara.soul.examples.dubbo.api.service.DubboTestService" uniqueServiceName="org.dromara.soul.examples.dubbo.api.service.DubboTestService" interface="org.dromara.soul.examples.dubbo.api.service.DubboTestService" generic="true" generic="true" sticky="false" timeout="10000" retries="2" loadbalance="random" id="org.dromara.soul.examples.dubbo.api.service.DubboTestService" valid="true" />

代码中genericService对象信息如下，他是一个代理对象，接口信息是从zookeeper中拿到的（在项目演示时用zookeeper作为注册中心）。

invoker :interface org.apache.dubbo.rpc.service.GenericService -> zookeeper://localhost:2181/org.apache.dubbo.registry.RegistryService?anyhost=true&application=soul_proxy&check=false&deprecated=false&dubbo=2.0.2&dynamic=true&generic=true&interface=org.dromara.soul.examples.dubbo.api.service.DubboTestService&loadbalance=random&methods=findById,insert,findAll&pid=22548&protocol=dubbo&register.ip=192.168.236.60&release=2.7.5&remote.application=test-dubbo-service&retries=2&side=consumer&sticky=false&timeout=10000&timestamp=1611918837429,directory: org.apache.dubbo.registry.integration.RegistryDirectory@46066629

真正的apache dubbo泛化调是genericService.$invokeAsync()，注意这里是异步的泛化调用。到这里，ApacheDubboPlugin插件的主要工作就完了，后面就是返回结果。

DubboResponsePlugin插件

这个插件就是对结果再一次包装，处理错误信息和成功的结果信息。

//org.dromara.soul.plugin.apache.dubbo.response.DubboResponsePlugin#execute    
@Override
    public Mono<Void> execute(final ServerWebExchange exchange, final SoulPluginChain chain) {
        return chain.execute(exchange).then(Mono.defer(() -> {
            final Object result = exchange.getAttribute(Constants.DUBBO_RPC_RESULT);
            //错误信息
            if (Objects.isNull(result)) {
                Object error = SoulResultWrap.error(SoulResultEnum.SERVICE_RESULT_ERROR.getCode(), SoulResultEnum.SERVICE_RESULT_ERROR.getMsg(), null);
                return WebFluxResultUtils.result(exchange, error);
            }
            //成功信息
            Object success = SoulResultWrap.success(SoulResultEnum.SUCCESS.getCode(), SoulResultEnum.SUCCESS.getMsg(), JsonUtils.removeClass(result));
            return WebFluxResultUtils.result(exchange, success);
        }));
    }

至此，就跟踪完了Soul网关中对Apache Dubbo插件处理的核心操作：接入dubbo服务，将http访问协议转化为dubbo协议，通过dubbo的泛化调用获取真正的接口服务信息。

Soul网关中的Http服务探活

28 Jan 2021   |   Soul  |  

服务探活机制是为了发现系统中上下游服务的的状态。当有新的服务注册时要通知其他系统，当有服务下线时也要告知其他系统。

Soul网关中有对Http服务处理的探活，所有的服务对象保存在soul-admin的UPSTREAM_MAP中，这里面的服务对象有两个来源，一个来自于原有的数据库，一个来自于其他服务的注册。

@Component
public class UpstreamCheckService {
	//保存上游服务
    private static final Map<String, List<DivideUpstream>> UPSTREAM_MAP = Maps.newConcurrentMap();
   
    //省略了其他代码
}

在UpstreamCheckService类中，在构造器执行完后，会执行setup()方法。里面做了两件事情：

  • 1.读取数据库的服务信息，保存到UPSTREAM_MAP；
  • 2.开启定时任务，检查每个方法。
    //在构造器只执行完，执行这个方法
	@PostConstruct
    public void setup() {
        PluginDO pluginDO = pluginMapper.selectByName(PluginEnum.DIVIDE.getName());
        if (pluginDO != null) {
		  //省略了其他代码
            //来源数据库的服务对象
                if (CollectionUtils.isNotEmpty(divideUpstreams)) {
                    UPSTREAM_MAP.put(selectorDO.getName(), divideUpstreams);
                }
            }
        }
        //是否开启探活机制，默认开启
        if (check) {
            //定时任务，每10秒执行一次
            new ScheduledThreadPoolExecutor(Runtime.getRuntime().availableProcessors(), SoulThreadFactory.create("scheduled-upstream-task", false))
                    .scheduleWithFixedDelay(this::scheduled, 10, scheduledTime, TimeUnit.SECONDS);
        }
    }

定时任务每10秒就去检查服务的状态，并且更新服务信息。

 private void scheduled() {
        if (UPSTREAM_MAP.size() > 0) {
            //检查每个选择器
            UPSTREAM_MAP.forEach(this::check);
        }
    }

    private void check(final String selectorName, final List<DivideUpstream> upstreamList) {
         //对每个服务进行检查
        for (DivideUpstream divideUpstream : upstreamList) {
            //检查服务
            final boolean pass = UpstreamCheckUtils.checkUrl(divideUpstream.getUpstreamUrl());
            if (pass) {
			//	成功
            } else {
                //失败
                divideUpstream.setStatus(false);
                log.error("check the url={} is fail ", divideUpstream.getUpstreamUrl());
            }
        }
        //都存活
        if (successList.size() == upstreamList.size()) {
            return;
        }
        //部分存活，只保留存活的，去除失活的服务
        if (successList.size() > 0) {
            UPSTREAM_MAP.put(selectorName, successList);
            updateSelectorHandler(selectorName, successList);
        } else { //没有存活的
            UPSTREAM_MAP.remove(selectorName);
            updateSelectorHandler(selectorName, null);
        }
    }

检查过程是通过socket进行连接，能够连接成功，则服务是好的，否则就认为服务连接失败。

    private static boolean isHostConnector(final String host, final int port) {
        try (Socket socket = new Socket()) {
            //建立socket连接
            socket.connect(new InetSocketAddress(host, port));
        } catch (IOException e) {
            return false;
        }
        return true;
    }

有服务失活的时候，要发送更新事件给网关。

    private void updateSelectorHandler(final String selectorName, final List<DivideUpstream> upstreams) {
        SelectorDO selectorDO = selectorMapper.selectByName(selectorName);
        if (Objects.nonNull(selectorDO)) {
			//省略了其他代码
                // publish change event.
                eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE, Collections.singletonList(selectorData)));
            }
        }
    }

当有新的服务启动时，SpringMvcClientBeanPostProcessor会处理接口信息，向soul-admin发起注册请求"/soul-client/springmvc-register"。当soul-admin端接收到这个请求时，会做三件事情：

  • 1.更新数据库信息；
  • 2.提交服务信息到UPSTREAM_MAP；
  • 3.发布事件到网关。
//接收  /soul-client/springmvc-register 请求的实现接口
public String registerSpringMvc(final SpringMvcRegisterDTO dto) {
        if (dto.isRegisterMetaData()) {
            MetaDataDO exist = metaDataMapper.findByPath(dto.getPath());
            if (Objects.isNull(exist)) {
                saveSpringMvcMetaData(dto);
            }
        }
        //处理请求
        String selectorId = handlerSpringMvcSelector(dto);
        handlerSpringMvcRule(selectorId, dto);
        return SoulResultMessage.SUCCESS;
   }

private String handlerSpringMvcSelector(final SpringMvcRegisterDTO dto) {
	//省略了其他代码
    
    // update db 更新数据库信息
    selectorMapper.updateSelective(selectorDO);
    // submit upstreamCheck 提交服务信息到UPSTREAM_MAP
    upstreamCheckService.submit(contextPath, addDivideUpstream);
    // publish change event. 发布事件到网关
    eventPublisher.publishEvent(new DataChangedEvent(ConfigGroupEnum.SELECTOR, DataEventTypeEnum.UPDATE,Collections.singletonList(selectorData)));

    }

上面的服务探活是在应用服务和soul-admin之间。其实，在网关也有一个UPSTREAM_MAP来保存服务信息。它里面的服务来源是soul-admin发布事件来通知网关的，网关会对信息进行处理，更新可用服务。

public final class UpstreamCacheManager {
		//保存服务信息
    private static final Map<String, List<DivideUpstream>> UPSTREAM_MAP = Maps.newConcurrentMap();
}

在网关这一边，服务信息处理流程是（假设数据同步采用的是websocket）：

  • SoulWebsocketClient: 后台 wesocket 信息在这里被监听，并发送给 WebsocketDataHandler 处理；
  • WebsocketDataHandler: 根据事件类型, 选择对应处理器 (PluginDataHandler、RuleDataHandler等)；
  • AbstractDataHandler: 根据事件变动类型(refresh、update等)， 调用处理器对应方法, 具体实现类会调用到 CommonPluginDataSubscriber 订阅器；
  • CommonPluginDataSubscriber: 这里存有所有注册为 Bean 的事件处理器，处理主要逻辑在subscribeDataHandler()方法中；
  • DividePluginDataHandler: 更新或移除缓存管理器中服务节点信息。

另外，在网关的UpstreamCacheManager中，也有一个每隔30秒的定时任务scheduled去检查服务的状态信息，但是这个默认是关闭的。

public final class UpstreamCacheManager {   
    
    //省略了其他代码
    
    private UpstreamCacheManager() {
            boolean check = Boolean.parseBoolean(System.getProperty("soul.upstream.check", "false"));
            if (check) {
                new ScheduledThreadPoolExecutor(1, SoulThreadFactory.create("scheduled-upstream-task", false))
                        .scheduleWithFixedDelay(this::scheduled,
                                30, Integer.parseInt(System.getProperty("soul.upstream.scheduledTime", "30")), TimeUnit.SECONDS);
            }
        }
}

小结，本篇文章介绍了在Soul中soul-admin和soul-bootstrap网关对服务的探活机制，主要实现类分别是UpstreamCheckService和UpstreamCacheManager。

Soul网关中的负载均衡

27 Jan 2021   |   Soul  |  

在divide插件中，Soul网关提供了负载均衡算法，对请求网关的IP选择一个真实的服务。在Soul中，负载均衡算法有三种：HashLoadBalance，RandomLoadBalance，RoundRobinLoadBalance。默认使用的是RandomLoadBalance算法。

使用时机

在执行divide插件时，会调用负载均衡算法，根据选择的结果，设置真实的请求服务。

//org.dromara.soul.plugin.divide.DividePlugin#doExecute
protected Mono<Void> doExecute(final ServerWebExchange exchange, final SoulPluginChain chain, final SelectorData selector, final RuleData rule) {
	
    	//省略了其他代码
    
        final String ip = Objects.requireNonNull(exchange.getRequest().getRemoteAddress()).getAddress().getHostAddress();
    	//使用负载均衡    
       DivideUpstream divideUpstream = LoadBalanceUtils.selector(upstreamList, ruleHandle.getLoadBalance(), ip);
      
        // 设置真实的请求服务
        String domain = buildDomain(divideUpstream);
      
   		 //省略了其他代码
        return chain.execute(exchange);
    }
LoadBalance的继承关系如下

类的设计使用的时模板方法设计模式：在抽象类中实现每个组件通用的功能，一些具体的功能留给子类去实现。在这里AbstrctLoadBalance实现了负载均衡的通用方法：获取权重，入参判断等。doSelect()留给了子类去实现，即具体的负载均衡算法逻辑。

public abstract class AbstractLoadBalance implements LoadBalance {
	
    //子类去实现
    protected abstract DivideUpstream doSelect(List<DivideUpstream> upstreamList, String ip);

    @Override
    public DivideUpstream select(final List<DivideUpstream> upstreamList, final String ip) {
        if (CollectionUtils.isEmpty(upstreamList)) {
            return null;
        }
        if (upstreamList.size() == 1) {
            return upstreamList.get(0);
        }
        return doSelect(upstreamList, ip);
    }

    protected int getWeight(final DivideUpstream upstream) {
     //省略了代码的具体实现
    }

	//省略其他代码
}

在select()方法中，先判断有没有服务；然后判断是否只有一个，是的话，就直接返回。因为一个不需要负载均衡，只能请求它。有多个就会通过负载均衡选择具体的类。

HashLoadBalance的原理

通过哈希算法计算所有服务地址，保存到map。然后也对当前请求的IP计算hash值，根据这个值到map中获取服务。

    @Override
    public DivideUpstream doSelect(final List<DivideUpstream> upstreamList, final String ip) {
        final ConcurrentSkipListMap<Long, DivideUpstream> treeMap = new ConcurrentSkipListMap<>();
       	//计算每个服务的hash
        for (DivideUpstream address : upstreamList) {
            for (int i = 0; i < VIRTUAL_NODE_NUM; i++) {
                long addressHash = hash("SOUL-" + address.getUpstreamUrl() + "-HASH-" + i);
                treeMap.put(addressHash, address);
            }
        }
        //计算当前的hash
        long hash = hash(String.valueOf(ip));
        //获取服务
        SortedMap<Long, DivideUpstream> lastRing = treeMap.tailMap(hash);
        if (!lastRing.isEmpty()) {
            return lastRing.get(lastRing.firstKey());
        }
        return treeMap.firstEntry().getValue();
    }
RandomLoadBalance的原理

首先，计算所有服务的权重，然后判断每个服务的权重是否相同，如果权重都相等或者为0，那么就随机选择一个。否则，会先获取到一个所有权重范围内的随机数，然后进行遍历，每次减去当前服务的权重值，如果小于0则返回结果。这里就会倾向于权重较大的那个服务。

    @Override
    public DivideUpstream doSelect(final List<DivideUpstream> upstreamList, final String ip) {
        //所有权重
        int totalWeight = calculateTotalWeight(upstreamList);
        //是否相同
        boolean sameWeight = isAllUpStreamSameWeight(upstreamList);
        if (totalWeight > 0 && !sameWeight) {
            return random(totalWeight, upstreamList);
        }
        // If the weights are the same or the weights are 0 then random
        return random(upstreamList);
    }
RoundRobinLoadBalance的原理

这个实现比较复杂，完整代码逻辑请看源码。核心思想： 对于每个服务维护一个权重对象，每次遍历所有权重对象获取到权重最大的那个服务，同时减去权重，以降低优先级。

@Override
    public DivideUpstream doSelect(final List<DivideUpstream> upstreamList, final String ip) {
        String key = upstreamList.get(0).getUpstreamUrl();
        ConcurrentMap<String, WeightedRoundRobin> map = methodWeightMap.get(key);

        for (DivideUpstream upstream : upstreamList) {
            String rKey = upstream.getUpstreamUrl();
  			
            //省略了其他代码
            
            //选择最大的
            if (cur > maxCurrent) {
                maxCurrent = cur;
                selectedInvoker = upstream;
                selectedWRR = weightedRoundRobin;
            }
            totalWeight += weight;
        }
        //省略了其他代码
        
        if (selectedInvoker != null) {
            selectedWRR.sel(totalWeight);
            return selectedInvoker;
        }
        // should not happen here
        return upstreamList.get(0);
    }

小结，主要讲解了Soul网关中的负载均衡实现原理。