Dubbo灰度发布课题研究

痛定思痛。 2023-07-03 06:23 25阅读 0赞

之前公司内部组织进行课题小组研究，我们组选择了DUBBO的灰度发布  
任务到手，就是分析，灰度发布时，我们要注意哪些？

一共5个注意点：  
1.数据库方面：DDL语句提前1天在业务低峰期由DBA进行操作，SQL代码层面，要检查，不允许出现select \* from table 类似语句，要保证表字段可以随时添加

2.对外流量灰度请求判定：![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMTQ0ODYx_size_16_color_FFFFFF_t_70]  
这里我们的设计是，流量进来以后，在网关层进行拦截器处理interceptor，网关层将请求信息转发到灰度策略服务，判定当前请求是否需要灰度，若需要灰度则在threadLocal中打上灰度标签is\_beta=true。这里我们也依靠了nacos来实现网关层灰度检查开关的动态配置。

3.dubbo-rpc接口灰度：  
3.1首先是dubbo-provider声明成beta模式，这个在provider启动前，可以通过添加自定义参数实现

<dubbo:service ref="userRpc" interface="com.xxx.user.api.IUserRpc" protocol="dubbo"
                   version="${dubbo.service.version}">
        <dubbo:parameter key="betaflag" value="${dubbo.provider.parameters.betaflag}"></dubbo:parameter>
    </dubbo:service>

这里通过添加参数betaflag参数，使该provider声明成beta模块，具体配置信息也是通过nacos管理

3.2dubbo路由策略修改，这里我自定义了一个BetaLoadBalance

package com.xxx.dubbo.balance;
    
    import com.alibaba.dubbo.common.URL;
    import com.alibaba.dubbo.rpc.Invocation;
    import com.alibaba.dubbo.rpc.Invoker;
    import com.alibaba.dubbo.rpc.cluster.loadbalance.AbstractLoadBalance;
    import com.xxx.pay.common.config.CommonConfig;
    import com.xxx.pay.common.util.SpringBeanTool;
    import com.xxx.pay.common.util.logback.BetaFlagUtils;
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.*;
    
    @Slf4j
    public class BetaLoadBalance extends AbstractLoadBalance  {
    
         static final String DUBBO_PROD_FlAG = "prod";
    
         static final String DUBBO_BETA_FlAG = "beta";
    
        public static final String NAME = "random";
    
        private final Random random = new Random();
    
        @Override
        protected <T> Invoker<T> doSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
    
            CommonConfig commonConfig =  (CommonConfig) SpringBeanTool.getBean("commonConfig");
    
            if(commonConfig == null){
                log.error("commonConfig对象为 null ");
            }else{
                log.info("commonConfig内容信息为：{} ",commonConfig.isIs_beta_mode());
            }
    
            if(commonConfig.isIs_beta_mode()){
                return this.betaSelect(invokers, url, invocation);
            }else{
                return this.randomSelect(invokers, url, invocation);
            }
    
    
        }
    
        /**
         *  写一个灰度路由策略
         * @param invokers
         * @param url
         * @param invocation
         * @param <T>
         * @return
         */
        private <T> Invoker<T> betaSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
            List<Invoker<T>> list = new ArrayList<>();
            for (Invoker invoker : invokers) {
                list.add(invoker);
            }
            Iterator<Invoker<T>> iterator = list.iterator();
            while (iterator.hasNext()) {
                Invoker<T> invoker = iterator.next();
                String providerStatus = invoker.getUrl().getParameter("betaflag", DUBBO_PROD_FlAG);
    
                log.info("轮询到的提供者为：{}, IP为 {} ",providerStatus,invoker.getUrl().getIp());
                //判断当前提供者，是否为灰度提供者
                if (Objects.equals(providerStatus, DUBBO_BETA_FlAG)) {
                    if(BetaFlagUtils.isBetaFlag()){
                        //判断当前线程threadLocal是否为灰度线程，灰度线程，直接返回beta提供者
                        log.info("灰度线程---返回灰度invoker");
                        return invoker;
                    }else {
                        //非灰度线程，移除beta提供者，避免实际用户流量到灰度服务
                        log.info("非灰度线程---移除灰度invoker");
                        iterator.remove();
                    }
                }else{
                    if(BetaFlagUtils.isBetaFlag()){
                       //非灰度提供者，先留着
                        continue;
                    }
                }
            }
    
            return this.randomSelect(list, url, invocation);
    
        }
    
    
        /**
         * 重写了一遍随机负载策略
         *
         * @param invokers
         * @param url
         * @param invocation
         * @param <T>
         * @return
         */
        private <T> Invoker<T> randomSelect(List<Invoker<T>> invokers, URL url, Invocation invocation) {
            int length = invokers.size(); // 总个数
            int totalWeight = 0; // 总权重
            boolean sameWeight = true; // 权重是否都一样
            for (int i = 0; i < length; i++) {
                int weight = getWeight(invokers.get(i), invocation);
                totalWeight += weight; // 累计总权重
                if (sameWeight && i > 0
                        && weight != getWeight(invokers.get(i - 1), invocation)) {
                    sameWeight = false; // 计算所有权重是否一样
                }
            }
            if (totalWeight > 0 && ! sameWeight) {
                // 如果权重不相同且权重大于0则按总权重数随机
                int offset = random.nextInt(totalWeight);
                // 并确定随机值落在哪个片断上
                for (int i = 0; i < length; i++) {
                    offset -= getWeight(invokers.get(i), invocation);
                    if (offset < 0) {
                        return invokers.get(i);
                    }
                }
            }
            // 如果权重相同或权重为0则均等随机
            return invokers.get(random.nextInt(length));
        }
        
        public BetaLoadBalance() {
        }
    
    }

这里的路由策略一个目的：  
**保证非灰度线程流量不到灰度提供者**  
3.3灰度标签传递：  
单应用场景是存储在threadLocal中的，不同应用之间该如何传递，这里我又扩展了dubbo的一个类

package com.xxx.pay.common.util.logback;
    
    import com.alibaba.dubbo.common.Constants;
    import com.alibaba.dubbo.common.extension.Activate;
    import com.alibaba.dubbo.rpc.*;
    import com.xxx.pay.common.logger.TpspLogger;
    import com.xxx.pay.common.util.TimeUtils;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.RandomStringUtils;
    import org.apache.commons.lang3.StringUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.slf4j.MDC;
    
    @Activate(group = { Constants.CONSUMER, Constants.PROVIDER })
    @Slf4j
    public class TraceIDFilter implements Filter {
       public static final String TRACE_ID = "TRACE_ID";
       public static final String BETA_FLAG = "BETA_FLAG";
    
       @Override
       public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
    
          //设置traceId日志链路跟踪
          setTraceId(invocation);
    
          //设置灰度标签，用于自定义路由策略
          String betaFlag = invocation.getAttachment(BETA_FLAG);
          if(StringUtils.isNotBlank(betaFlag)){
             BetaFlagUtils.setBetaFlag(betaFlag);
          }else {
             BetaFlagUtils.setBetaFlag(BetaFlagUtils.isBetaFlag());
          }
          invocation.getAttachments().put(BETA_FLAG, BetaFlagUtils.getBetaFlag());
          Result result = invoker.invoke(invocation);
          return result;
       }
    
       private void setTraceId(Invocation invocation){
          String mdcData = invocation.getAttachment(TRACE_ID);
          if (mdcData != null) {
             TraceIDUtils.setTraceId(mdcData);
             MDC.put(TraceIDUtils.LOG_KEY, mdcData);
          } else if (TraceIDUtils.getTraceId() != null) {
             mdcData = TraceIDUtils.getTraceId();
             invocation.getAttachments().put(TRACE_ID, mdcData);
             MDC.put(TraceIDUtils.LOG_KEY, mdcData);
          } else {
             TraceIDUtils.setMCData(TimeUtils.getCurrent() + RandomStringUtils.randomNumeric(10));
             mdcData = TraceIDUtils.getTraceId();
             invocation.getAttachments().put(TRACE_ID, mdcData);
          }
       }
    
    }

通过扩展dubbo的filter来实现，这个类传递了链路跟踪号traceId(用于日志)、betaFlag(用于灰度)

4.rocketMQ的灰度传递  
塞消息时，塞一个参数进去beta\_flag。然后消费者根据消息体中的beta\_flag就可以调用不同的RPC服务，实现灰度

5.定时任务  
这里引入了XXL-job

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMTQ0ODYx_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200203154431277.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMzMTQ0ODYx,size_16,color_FFFFFF,t_70