记线上Dubbo调用异常排查处理

@Activate(group = Constants.CONSUMER)public class HystrixFilter implements Filter {
    public HystrixFilter() {        ApplicationContext springContext = ApplicationContextHolder.getContext();        if (springContext != null && !springContext.containsBean(HystrixSpringService.class.getSimpleName())) {            BeanFactory beanFactory = springContext.getAutowireCapableBeanFactory();            if (beanFactory instanceof DefaultListableBeanFactory) {                BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.rootBeanDefinition(HystrixSpringService.class);                beanDefinitionBuilder.setDestroyMethodName("preDestroy");                beanDefinitionBuilder.setScope("singleton");
                ((DefaultListableBeanFactory) beanFactory).registerBeanDefinition(HystrixSpringService.class.getSimpleName(), beanDefinitionBuilder.getBeanDefinition());
                //触发初始化                beanFactory.getBean(HystrixSpringService.class.getSimpleName());            }        }    }
    @Override    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {        // 异步调用使用hystrix做熔断没有意义        if ("true".equalsIgnoreCase(invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY))) {            return invoker.invoke(invocation);        }
        DubboHystrixCommand command = new DubboHystrixCommand(invoker, invocation);        return command.execute();    }
}

public class DubboHystrixCommand extends HystrixCommand<Result> {
    private static Logger logger = Logger.getLogger(DubboHystrixCommand.class);    private static final int DEFAULT_THREADPOOL_CORE_SIZE = 30;    private static final int CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS = 30000;    private static final int DEFAULT_FAIL_QPS_THRESHOLD = 20;    private Invoker invoker;    private Invocation invocation;
    public DubboHystrixCommand(Invoker<?> invoker, Invocation invocation) {        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(invoker.getInterface().getName()))                .andCommandKey(HystrixCommandKey.Factory.asKey(String.format("%s_%d", invocation.getMethodName(),                        invocation.getArguments() == null ? 0 : invocation.getArguments().length)))                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()                        //10秒钟内请求失败上限值，超过此值熔断器发挥作用                        .withCircuitBreakerRequestVolumeThreshold(getFailQpsThreshold(invoker.getUrl()) * 10)                        //熔断器中断请求30秒后会进入半打开状态,放部分流量过去重试                        .withCircuitBreakerSleepWindowInMilliseconds(getCircuitBreakerSleepWindowInMilliseconds(invoker.getUrl()))                        //错误率达到50开启熔断保护                        .withCircuitBreakerErrorThresholdPercentage(50)                        //使用dubbo的超时，禁用这里的超时                        .withExecutionTimeoutEnabled(false))                //根据dubbo配置设置线程池大小                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(getThreadPoolCoreSize(invoker.getUrl()))));

        this.invoker = invoker;        this.invocation = invocation;    }
    /**     * 获取每秒请求失败的阈值，超过此阈值熔断器开始生效     *     * @param url     * @return     */    private static int getFailQpsThreshold(URL url) {        if (url != null) {            int threshold = url.getParameter("FailQpsThreshold", DEFAULT_FAIL_QPS_THRESHOLD);            if (logger.isDebugEnabled()) {                logger.debug("FailQpsThreshold: " + threshold);            }            return threshold;        }
        return DEFAULT_FAIL_QPS_THRESHOLD;    }
    /**     * 获取熔断器中断请求窗口大小     *     * @param url     * @return     */    private static int getCircuitBreakerSleepWindowInMilliseconds(URL url) {        if (url != null) {            int circuitBreakerSleepWindowInMilliseconds = url.getParameter("CircuitBreakerSleepWindowInMilliseconds", CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS);            if (logger.isDebugEnabled()) {                logger.debug("circuitBreakerSleepWindowInMilliseconds: " + circuitBreakerSleepWindowInMilliseconds);            }            return circuitBreakerSleepWindowInMilliseconds;        }
        return CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS;
    }
    /**     * 获取线程池大小     *     * @param url     * @return     */    private static int getThreadPoolCoreSize(URL url) {        if (url != null) {            int size = url.getParameter("ThreadPoolCoreSize", DEFAULT_THREADPOOL_CORE_SIZE);            if (logger.isDebugEnabled()) {                logger.debug("ThreadPoolCoreSize: " + size);            }            return size;        }
        return DEFAULT_THREADPOOL_CORE_SIZE;
    }
    @Override    protected Result run() throws Exception {        Throwable exception = null;
        Result result = null;
        try {            result = invoker.invoke(invocation);            exception = result.getException();        } catch (Exception e) {            exception = e;        }
        // 这里打印异常是为了记录异常，再抛出异常是为了触发fallback        if (exception != null) {            Logs.error("Dubbo Exception: ", exception);            throw new Exception(exception);        }
        return result;    }
    @Override    protected Result getFallback() {        return new RpcResult((Object) null);    }}public class DubboHystrixCommand extends HystrixCommand<Result> {
    private static Logger logger = Logger.getLogger(DubboHystrixCommand.class);    private static final int DEFAULT_THREADPOOL_CORE_SIZE = 30;    private static final int CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS = 30000;    private static final int DEFAULT_FAIL_QPS_THRESHOLD = 20;    private Invoker invoker;    private Invocation invocation;
    public DubboHystrixCommand(Invoker<?> invoker, Invocation invocation) {        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(invoker.getInterface().getName()))                .andCommandKey(HystrixCommandKey.Factory.asKey(String.format("%s_%d", invocation.getMethodName(),                        invocation.getArguments() == null ? 0 : invocation.getArguments().length)))                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()                        //10秒钟内请求失败上限值，超过此值熔断器发挥作用                        .withCircuitBreakerRequestVolumeThreshold(getFailQpsThreshold(invoker.getUrl()) * 10)                        //熔断器中断请求30秒后会进入半打开状态,放部分流量过去重试                        .withCircuitBreakerSleepWindowInMilliseconds(getCircuitBreakerSleepWindowInMilliseconds(invoker.getUrl()))                        //错误率达到50开启熔断保护                        .withCircuitBreakerErrorThresholdPercentage(50)                        //使用dubbo的超时，禁用这里的超时                        .withExecutionTimeoutEnabled(false))                //根据dubbo配置设置线程池大小                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter().withCoreSize(getThreadPoolCoreSize(invoker.getUrl()))));

        this.invoker = invoker;        this.invocation = invocation;    }
    /**     * 获取每秒请求失败的阈值，超过此阈值熔断器开始生效     *     * @param url     * @return     */    private static int getFailQpsThreshold(URL url) {        if (url != null) {            int threshold = url.getParameter("FailQpsThreshold", DEFAULT_FAIL_QPS_THRESHOLD);            if (logger.isDebugEnabled()) {                logger.debug("FailQpsThreshold: " + threshold);            }            return threshold;        }
        return DEFAULT_FAIL_QPS_THRESHOLD;    }
    /**     * 获取熔断器中断请求窗口大小     *     * @param url     * @return     */    private static int getCircuitBreakerSleepWindowInMilliseconds(URL url) {        if (url != null) {            int circuitBreakerSleepWindowInMilliseconds = url.getParameter("CircuitBreakerSleepWindowInMilliseconds", CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS);            if (logger.isDebugEnabled()) {                logger.debug("circuitBreakerSleepWindowInMilliseconds: " + circuitBreakerSleepWindowInMilliseconds);            }            return circuitBreakerSleepWindowInMilliseconds;        }
        return CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS;
    }
    /**     * 获取线程池大小     *     * @param url     * @return     */    private static int getThreadPoolCoreSize(URL url) {        if (url != null) {            int size = url.getParameter("ThreadPoolCoreSize", DEFAULT_THREADPOOL_CORE_SIZE);            if (logger.isDebugEnabled()) {                logger.debug("ThreadPoolCoreSize: " + size);            }            return size;        }
        return DEFAULT_THREADPOOL_CORE_SIZE;
    }
    @Override    protected Result run() throws Exception {        Throwable exception = null;
        Result result = null;
        try {            result = invoker.invoke(invocation);            exception = result.getException();        } catch (Exception e) {            exception = e;        }
        // 这里打印异常是为了记录异常，再抛出异常是为了触发fallback        if (exception != null) {            Logs.error("Dubbo Exception: ", exception);            throw new Exception(exception);        }
        return result;    }
    @Override    protected Result getFallback() {        return new RpcResult((Object) null);    }}

@Activate(group = Constants.CONSUMER)public class HystrixFilter implements Filter {
    public HystrixFilter() {        ApplicationContext springContext = ApplicationContextHolder.getContext();        if (springContext != null && !springContext.containsBean(HystrixSpringService.class.getSimpleName())) {            BeanFactory beanFactory = springContext.getAutowireCapableBeanFactory();            if (beanFactory instanceof DefaultListableBeanFactory) {                BeanDefinitionBuilder beanDefinitionBuilder = BeanDefinitionBuilder.rootBeanDefinition(HystrixSpringService.class);                beanDefinitionBuilder.setDestroyMethodName("preDestroy");                beanDefinitionBuilder.setScope("singleton");
                ((DefaultListableBeanFactory) beanFactory).registerBeanDefinition(HystrixSpringService.class.getSimpleName(), beanDefinitionBuilder.getBeanDefinition());
                //触发初始化                beanFactory.getBean(HystrixSpringService.class.getSimpleName());            }        }    }
    @Override    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {        // 异步调用使用hystrix做熔断没有意义        if ("true".equalsIgnoreCase(invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY))                || "false".equalsIgnoreCase(invoker.getUrl().getMethodParameter(invocation.getMethodName(), "hystrixOpen"))) {            return invoker.invoke(invocation);        }
        DubboHystrixCommand command = new DubboHystrixCommand(invoker, invocation);        return command.execute();    }
}

public class DubboHystrixCommand extends HystrixCommand<Result> {
    private static Logger logger = Logger.getLogger(DubboHystrixCommand.class);    private static final int DEFAULT_THREAD_POOL_CORE_SIZE = 30;    private static final int DEFAULT_THREAD_POOL_MAX_SIZE = 50;    private static final int DEFAULT_THREAD_POOL_KEEP_ALIVE_TIME_MINUTES = 5;
    private static final int DEFAULT_FAIL_QPS_THRESHOLD = 20;    private Invoker invoker;    private Invocation invocation;
    public DubboHystrixCommand(Invoker<?> invoker, Invocation invocation) {        super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(invoker.getInterface().getName()))                .andCommandKey(HystrixCommandKey.Factory.asKey(String.format("%s_%d", invocation.getMethodName(),                        invocation.getArguments() == null ? 0 : invocation.getArguments().length)))                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()                        //10秒钟内请求失败上限值，超过此值熔断器发挥作用                        .withCircuitBreakerRequestVolumeThreshold(getFailQpsThreshold(invoker.getUrl()) * 10)                        //熔断器中断请求30秒后会进入半打开状态,放部分流量过去重试                        .withCircuitBreakerSleepWindowInMilliseconds(30000)                        //错误率达到50开启熔断保护                        .withCircuitBreakerErrorThresholdPercentage(50)                        //使用dubbo的超时，禁用这里的超时                        .withExecutionTimeoutEnabled(false))                //根据dubbo配置设置线程池大小                .andThreadPoolPropertiesDefaults(getThreadPoolSetter(invoker.getUrl())));

        this.invoker = invoker;        this.invocation = invocation;    }
    /**     * 获取每秒请求失败的阈值，超过此阈值熔断器开始生效     *     * @param url     * @return     */    private static int getFailQpsThreshold(URL url) {        if (url != null) {            int threshold = url.getParameter("FailQpsThreshold", DEFAULT_FAIL_QPS_THRESHOLD);            if (logger.isDebugEnabled()) {                logger.debug("FailQpsThreshold: " + threshold);            }            return threshold;        }
        return DEFAULT_FAIL_QPS_THRESHOLD;    }
    private static HystrixThreadPoolProperties.Setter getThreadPoolSetter(URL url) {        return HystrixThreadPoolProperties.Setter()                .withCoreSize(getThreadPoolProperties(url, "threadPoolCoreSize", DEFAULT_THREAD_POOL_CORE_SIZE))                .withMaximumSize(getThreadPoolProperties(url, "threadPoolMaxSize", DEFAULT_THREAD_POOL_MAX_SIZE))                .withKeepAliveTimeMinutes(getThreadPoolProperties(url, "threadPoolKeepAliveTimeMinutes",                        DEFAULT_THREAD_POOL_KEEP_ALIVE_TIME_MINUTES));    }
    /**     * 获取线程池大小     *     * @param url     * @return     */    private static int getThreadPoolProperties(URL url, String name, int defaultProperties) {        if (url != null) {            int size = url.getParameter(name, defaultProperties);            if (logger.isDebugEnabled()) {                logger.debug(name + ": " + size);            }            return size;        }        return defaultProperties;    }
    @Override    protected Result run() throws Exception {        Throwable exception;
        Result result = null;        try {            result = invoker.invoke(invocation);            exception = result.getException();        } catch (Exception e) {            exception = e;        }
        // 有异常抛出        if (exception != null) {            Logs.error("dubbo exception: ", exception);            throw new RuntimeException(exception);        }
        return result;    }
}