一、 Dubbo的SPI中的Adaptive功能（自动适配功能）

　　Dubbo中的Adaptive功能，主要解决的问题是如何动态的选择具体的扩展点。通过getAdaptiveExtension 统一对指定接口对应的所有扩展点进行封装，通过URL的方式对扩展点来进行

动态选择。 (dubbo中所有的注册信息都是通过URL的形式进行处理的。)这里同样采用相同的方式进行实现。

（1）创建接口

　　api中的 HelloService 扩展如下方法,在sayHello方法上增加 @Adaptive 注解，并且在参数中提供URL参数.注意这里的URL参数的类为 org.apache.dubbo.common.URL

其中@SP可以指定一个字符串参数，用于指明该SPI的默认实现。

package city.albert;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Adaptive;
import org.apache.dubbo.common.extension.SPI;

/**
 * @author niunafei
 * @function
 * @email niunafei0315@163.com
 * @date 2020/8/26  11:08 AM
 */
@SPI("default")
public interface HelloService {

    /**
     * 抽象方法
     *
     * @param name
     * @return
     */
    String sayHello(String name);


    /**
     * 抽象方法
     *
     * @param url
     * @param name
     * @return
     */
    @Adaptive
    String sayHello(URL url, String name);
}

View Code

（2）创建实现类

在Service实现类中实现接口，这里为了区分创建了两个类

package city.albert.impl;

import city.albert.HelloService;
import org.apache.dubbo.common.URL;

/**
 * @author niunafei
 * @function
 * @email niunafei0315@163.com
 * @date 2020/8/26  11:11 AM
 */
public class DefaultServiceImpl implements HelloService{
    @Override
    public String sayHello(String name) {
        return "默认输出：你好! "+name;
    }

    @Override
    public String sayHello(URL url, String name) {
        return "默认输出：你好! "+name;
    }
}

View Code

package city.albert.impl;

import city.albert.HelloService;
import org.apache.dubbo.common.URL;

/**
 * @author niunafei
 * @function
 * @email niunafei0315@163.com
 * @date 2020/8/26  11:16 AM
 */
public class UserServiceImpl implements HelloService {
    @Override
    public String sayHello(String name) {
        return "自定义输出：你好! " + name;
    }

    @Override
    public String sayHello(URL url, String name) {
        return "自定义输出：你好! " + name;
    }
}

View Code

（3）在META-INF/dubbo目录下创建文件名为接口city.albert.HelloService的文件，内容如下

default=city.albert.impl.DefaultServiceImpl
user=city.albert.impl.UserServiceImpl

（4）编写MainTest

最后在获取的时候方式有所改变，需要传入URL参数，并且在参数中指定具体的实现类参数

如:

package city.albert;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.ExtensionLoader;

import java.util.List;

/**
 * @author niunafei
 * @function
 * @email niunafei0315@163.com
 * @date 2020/8/26  11:19 AM
 */
public class MainTest {

    public static void main(String[] args) {
        URL url=URL.valueOf("test://localhost/test?hello.service=user");
        HelloService adaptiveExtension = ExtensionLoader.getExtensionLoader(HelloService.class).getAdaptiveExtension();

        System.out.println(adaptiveExtension.sayHello(url,"张三"));

        System.out.println(
                adaptiveExtension.sayHello(URL.valueOf("test://localhost/test"),"张三")
        );
    }
}

View Code

注意：

因为在这里只是临时测试，协议与地址均是不存在的，关键的点在于hello.service 参数，这个参数的值指定的就是具体的实现方式。为什么叫hello.service 是因为这个接口的名称，其中后面的大写部分被dubbo自动转码为 . 分割。通过 getAdaptiveExtension 来提供一个统一的类来对所有的扩展点提供支持(底层对所有的扩展点进行封装)。

调用时通过参数中增加 URL 对象来实现动态的扩展点使用。

如果URL没有提供该参数，则该方法会使用默认在 SPI 注解中声明的实现。

二、Dubbo调用时拦截操作

与很多框架一样，Dubbo也存在拦截（过滤）机制，可以通过该机制在执行目标程序前后执行我们指定

的代码。

Dubbo的Filter机制，是专门为服务提供方和服务消费方调用过程进行拦截设计的，每次远程方法执行，该拦截都会被执行。这样就为开发者提供了非常方便的扩展性，比如为dubbo接口实现ip白名单功

能、监控功能、日志记录等。

步骤如下：

（1）实现 org.apache.dubbo.rpc.Filter 接口

（2）使用 org.apache.dubbo.common.extension.Activate 接口进行对类进行注册通过group 可以指定生产端消费端如:@Activate(group = {CommonConstants.CONSUMER})

（3）计算方法运行时间的代码实现在 META-INF.dubbo 中新建 org.apache.dubbo.rpc.Filter 文件，并将当前类的全名写入

注意：一般类似于这样的功能都是单独开发依赖的，所以再使用方的项目中只需要引入依赖，在调用接口时，该方法便会自动拦截。

自定义filter响应时间记录如下：

package city.albert.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

/**
 * @author niunafei
 * @function
 * @email niunafei0315@163.com
 * @date 2020/8/26  12:23 PM
 * <p>
 * <p>
 * 指定在消费端拦截
 */
@Activate(group = {CommonConstants.CONSUMER})
public class TestFilter implements Filter {
    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        Long time = System.currentTimeMillis();
        //执行方法
        Result result = invoker.invoke(invocation);
        System.out.println("日志时间 time" + (System.currentTimeMillis() - time));
        return result;
    }
}

三、负载均衡策略

负载均衡（Load Balance）, 其实就是将请求分摊到多个操作单元上进行执行，从而共同完成工作任务。负载均衡策略主要用于客户端存在多个提供者时进行选择某个提供者。在集群负载均衡时，Dubbo 提供了多种均衡策略（包括随机、轮询、最少活跃调用数、一致性Hash），缺省为random随机调用。

配置负载均衡策略，既可以在服务提供者一方配置，也可以在服务消费者一方配置，如下：

//在服务消费者一方配置负载均衡策略 
@Reference(check = false,loadbalance = "random")

//在服务提供者一方配置负载均衡
@Service(loadbalance = "random") 
public class HelloServiceImpl implements HelloService { 
    public String sayHello(String name) {
         return "hello " + name; 
    }
 }

自定义如下：

负载均衡器在Dubbo中的SPI接口是 org.apache.dubbo.rpc.cluster.LoadBalance , 可以通过实现这个接口来实现自定义的负载均衡规则。

（1）自定义负载均衡器

创建负载均衡器OnlyFirstLoadbalancer。这里功能只是简单的选取所有机器中的第一个(按照字母排序 + 端口排序)。

（2）配置负载均衡器

在dubbo-spi-loadbalance工程的 META-INF/dubbo 目录下新建org.apache.dubbo.rpc.cluster.LoadBalance 文件，并将当前类的全名写入:形式 key=value

onlyFirst=包名.负载均衡器

（3）在服务提供者工程实现类中编写用于测试负载均衡效果的方法启动不同端口时方法返回的信息不同

（4）启动多个服务要求他们使用同一个接口注册到同一个注册中心但是他们的dubbo通信端口不同

（5）在服务消费方指定自定义负载均衡器 onlyFirst

@Reference(loadbalance = "onlyFirst")

（6）测试自定义负载均衡的效果

四、线程池

官方线程文档，dubbo在使用时，都是通过创建真实的业务线程池进行操作的。目前已知的线程池模型有两个和java中的相互对应:

1、fix: 表示创建固定大小的线程池。也是Dubbo默认的使用方式，默认创建的执行线程数为200，并且是没有任何等待队列的。所以再极端的情况下可能会存在问题，比如某个操作大量执行时，
可能存在堵塞的情况。后面也会讲相关的处理办法。
2、cache: 创建非固定大小的线程池，当线程不足时，会自动创建新的线程。但是使用这种的时候需要注意，如果突然有高TPS的请求过来，方法没有及时完成，则会造成大量的线程创建，对系统的
CPU和负载都是压力，执行越多反而会拖慢整个系统。

自定义线程池：

（1）线程池实现, 这里主要是基于对 FixedThreadPool 中的实现做扩展出线程监控的部分

package city.albert;

import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.threadpool.support.fixed.FixedThreadPool;

import java.util.Map;
import java.util.concurrent.*;

/**
 * @author niunafei
 * @function
 * @email niunafei0315@163.com
 * @date 2020/8/26  12:53 PM
 */
public class WatchingThreadPool extends FixedThreadPool implements Runnable {
    private static final double ALARM_PERCENT = 0.90;
    private final Map<URL, ThreadPoolExecutor> THREAD_POOLS = new ConcurrentHashMap<>();

    public WatchingThreadPool() {
        // 每隔3秒打印线程使用情况
        ScheduledFuture<?> schedule = Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(this, 1, 3, TimeUnit.SECONDS);

    }

    @Override
    public Executor getExecutor(URL url) {
        // 从父类中创建线程池
        final Executor executor = super.getExecutor(url);
        if (executor instanceof ThreadPoolExecutor) {
            THREAD_POOLS.put(url, ((ThreadPoolExecutor) executor));
        }
        return executor;
    }


    @Override
    public void run() {
        // 遍历线程池，如果超出指定的部分，进行操作，比如接入公司的告警系统或者短信平台
        for (Map.Entry<URL, ThreadPoolExecutor> entry : THREAD_POOLS.entrySet()) {
            final URL url = entry.getKey();
            final ThreadPoolExecutor executor = entry.getValue();
            // 当前执行中的线程数
            final int activeCount = executor.getActiveCount();
            // 总计线程数
            final int poolSize = executor.getCorePoolSize();
            double used = (double) activeCount / poolSize;
            final int usedNum = (int) (used * 100);
            System.out.println(String.format("线程池执行状态:[%d/%d]:%d", activeCount, poolSize, usedNum));
            if (used >= ALARM_PERCENT) {
                System.out.println(String.format("超出警戒值！host:%s, 当前已使用量:%d , URL:%s", url.getIp(), usedNum, url.toFullString()));
            }
        }
    }
}

View Code

（2）SPI声明，创建文件 META-INF/dubbo/org.apache.dubbo.common.threadpool.ThreadPool

watching=包名.线程池名

（3）在服务提供方项目引入该依赖

（4）在服务提供方项目中设置使用该线程池生成器

dubbo.provider.threadpool=watching

的提供者超过1秒的时间(比如这里用休眠 Thread.sleep )，消费者则需要启动多个线程来并行执行，来

模拟整个并发情况。

（6）在调用方则尝试简单通过for循环启动多个线程来执行查看服务提供方的监控情况

五、路由规则

路由规则官网具体详情可以参考官网教程。这里呢结合路由做出了一个上线系统的结合案例：灰度实现机器伸缩

实现主体思路

1.利用zookeeper的路径感知能力，在服务准备进行重启之前将当前机器的IP地址和应用名写入 zookeeper。 
2.服务消费者监听该目录，读取其中需要进行关闭的应用名和机器IP列表并且保存到内存中。
3.当前请求过来时，判断是否是请求该应用，如果是请求重启应用，则将该提供者从服务列表中移除。

（1）引入 Curator 框架，用于方便操作Zookeeper

（2）编写Zookeeper的操作类，用于方便进行zookeeper处理

（3）编写需要进行预发布的路径管理器，用于缓存和监听所有的待灰度机器信息列表。

（4）编写路由类(实现 org.apache.dubbo.rpc.cluster.Router )RestartingInstanceRouter，主要目的在于对ReadyRestartInstances 中的数据进行处理，并且移除路由调用列表中正在重启中的服务。

（5）由于 Router 机制比较特殊，所以需要利用一个专门的 RouterFactory 来生成，原因在于并不是所有的都需要添加路由，所以需要利用 @Activate 来锁定具体哪些服务才需要生成使用。

@Activate 
public class RestartingInstanceRouterFactory implements RouterFactory { 
    @Override 
    public Router getRouter(URL url) { 
        return new RestartingInstanceRouter(url); 
    } 
}

（6）对 RouterFactory 进行注册，同样放入到META-INF/dubbo/org.apache.dubbo.rpc.cluster.RouterFactory 文件中。

（7）将dubbo-spi-router项目引入至 consumer 项目的依赖中。

（8）这时直接启动程序，还是利用上面中所写好的 consumer 程序进行执行，确认各个 provider 可以正常执行。

（9）单独写一个 main 函数来进行将某台实例设置为启动中的状态，比如这里我们认定为当前这台机器中的 service-provider 这个提供者需要进行重启操作。

（10）执行完成后，再次进行尝试通过 consumer 进行调用，即可看到当前这台机器没有再发送任何请求

（11）一般情况下，当机器重启到一定时间后，我们可以再通过 removeRestartingInstance 方法对这个机器设定为既可以继续执行。

（12）调用完成后，我们再次通过 consumer 去调用，即可看到已经再次恢当前机器的请求参数。

主要应用zk的节点添加监听功能，进行添加ip+端口的节点，进行增删、监听处理

Dubbo高级进阶Spi应用

一、 Dubbo的SPI中的Adaptive功能（自动适配功能）

二、Dubbo调用时拦截操作

三、负载均衡策略

四、线程池

推荐阅读