diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java
index 74a4fe4..04bb400 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/constants/MetricsConstants.java
@@ -73,4 +73,6 @@
     int PROMETHEUS_DEFAULT_PUSH_INTERVAL = 30;
 
     String PROMETHEUS_DEFAULT_JOB_NAME = "default_dubbo_job";
+
+    String METRIC_FILTER_START_TIME = "metric_filter_start_time";
 }
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/DefaultMetricsCollector.java b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/DefaultMetricsCollector.java
index 599b6c8..6e5ba2d 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/DefaultMetricsCollector.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/DefaultMetricsCollector.java
@@ -17,27 +17,26 @@
 
 package org.apache.dubbo.common.metrics.collector;
 
-import org.apache.dubbo.common.metrics.event.MetricsEvent;
-import org.apache.dubbo.common.metrics.event.RTEvent;
+import static org.apache.dubbo.common.metrics.model.MetricsCategory.REQUESTS;
+import static org.apache.dubbo.common.metrics.model.MetricsCategory.RT;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.apache.dubbo.common.metrics.collector.stat.MetricsStatComposite;
+import org.apache.dubbo.common.metrics.collector.stat.MetricsStatHandler;
 import org.apache.dubbo.common.metrics.event.RequestEvent;
 import org.apache.dubbo.common.metrics.listener.MetricsListener;
-import org.apache.dubbo.common.metrics.model.MethodMetric;
 import org.apache.dubbo.common.metrics.model.MetricsKey;
 import org.apache.dubbo.common.metrics.model.sample.GaugeMetricSample;
 import org.apache.dubbo.common.metrics.model.sample.MetricSample;
 import org.apache.dubbo.rpc.model.ApplicationModel;
 
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.atomic.LongAccumulator;
-
-import static org.apache.dubbo.common.metrics.model.MetricsCategory.REQUESTS;
-import static org.apache.dubbo.common.metrics.model.MetricsCategory.RT;
-
 /**
  * Default implementation of {@link MetricsCollector}
  */
@@ -46,23 +45,11 @@
     private AtomicBoolean collectEnabled = new AtomicBoolean(false);
     private final List<MetricsListener> listeners = new ArrayList<>();
     private final ApplicationModel applicationModel;
-    private final String applicationName;
-
-    private final Map<MethodMetric, AtomicLong> totalRequests = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, AtomicLong> succeedRequests = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, AtomicLong> failedRequests = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, AtomicLong> processingRequests = new ConcurrentHashMap<>();
-
-    private final Map<MethodMetric, AtomicLong> lastRT = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, LongAccumulator> minRT = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, LongAccumulator> maxRT = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, AtomicLong> avgRT = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, AtomicLong> totalRT = new ConcurrentHashMap<>();
-    private final Map<MethodMetric, AtomicLong> rtCount = new ConcurrentHashMap<>();
+    private final MetricsStatComposite stats;
 
     public DefaultMetricsCollector(ApplicationModel applicationModel) {
         this.applicationModel = applicationModel;
-        this.applicationName = applicationModel.getApplicationName();
+        this.stats = new MetricsStatComposite(applicationModel.getApplicationName(), this);
     }
 
     public void setCollectEnabled(Boolean collectEnabled) {
@@ -77,81 +64,51 @@
         listeners.add(listener);
     }
 
-    public void increaseTotalRequests(String interfaceName, String methodName, String group, String version) {
-        if (isCollectEnabled()) {
-            MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
-            AtomicLong count = totalRequests.computeIfAbsent(metric, k -> new AtomicLong(0L));
-            count.incrementAndGet();
+    public List<MetricsListener> getListener() {
+        return this.listeners;
+    }
 
-            publishEvent(new RequestEvent(metric, RequestEvent.Type.TOTAL));
-        }
+    public void increaseTotalRequests(String interfaceName, String methodName, String group, String version) {
+        doExecute(RequestEvent.Type.TOTAL,statHandler-> {
+            statHandler.increase(interfaceName, methodName, group, version);
+        });
     }
 
     public void increaseSucceedRequests(String interfaceName, String methodName, String group, String version) {
-        if (isCollectEnabled()) {
-            MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
-            AtomicLong count = succeedRequests.computeIfAbsent(metric, k -> new AtomicLong(0L));
-            count.incrementAndGet();
-
-            publishEvent(new RequestEvent(metric, RequestEvent.Type.SUCCEED));
-        }
+        doExecute(RequestEvent.Type.SUCCEED,statHandler->{
+            statHandler.increase(interfaceName, methodName, group, version);
+        });
     }
 
-    public void increaseFailedRequests(String interfaceName, String methodName, String group, String version) {
-        if (isCollectEnabled()) {
-            MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
-            AtomicLong count = failedRequests.computeIfAbsent(metric, k -> new AtomicLong(0L));
-            count.incrementAndGet();
+    public void increaseFailedRequests(String interfaceName,
+                                       String methodName,
+                                       String group,
+                                       String version) {
+        doExecute(RequestEvent.Type.FAILED,statHandler->{
+            statHandler.increase(interfaceName, methodName, group, version);
+        });
+    }
 
-            publishEvent(new RequestEvent(metric, RequestEvent.Type.FAILED));
-        }
+    public void businessFailedRequests(String interfaceName, String methodName, String group, String version) {
+        doExecute(RequestEvent.Type.BUSINESS_FAILED,statHandler->{
+            statHandler.increase(interfaceName, methodName, group, version);
+        });
     }
 
     public void increaseProcessingRequests(String interfaceName, String methodName, String group, String version) {
-        if (isCollectEnabled()) {
-            MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
-            AtomicLong count = processingRequests.computeIfAbsent(metric, k -> new AtomicLong(0L));
-            count.incrementAndGet();
-        }
+        doExecute(RequestEvent.Type.PROCESSING,statHandler-> {
+            statHandler.increase(interfaceName, methodName, group, version);
+        });
     }
 
     public void decreaseProcessingRequests(String interfaceName, String methodName, String group, String version) {
-        if (isCollectEnabled()) {
-            MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
-            AtomicLong count = processingRequests.computeIfAbsent(metric, k -> new AtomicLong(0L));
-            count.decrementAndGet();
-        }
+        doExecute(RequestEvent.Type.PROCESSING,statHandler-> {
+            statHandler.decrease(interfaceName, methodName, group, version);
+        });
     }
 
     public void addRT(String interfaceName, String methodName, String group, String version, Long responseTime) {
-        if (isCollectEnabled()) {
-            MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
-
-            AtomicLong last = lastRT.computeIfAbsent(metric, k -> new AtomicLong());
-            last.set(responseTime);
-
-            LongAccumulator min = minRT.computeIfAbsent(metric, k -> new LongAccumulator(Long::min, Long.MAX_VALUE));
-            min.accumulate(responseTime);
-
-            LongAccumulator max = maxRT.computeIfAbsent(metric, k -> new LongAccumulator(Long::max, Long.MIN_VALUE));
-            max.accumulate(responseTime);
-
-            AtomicLong total = totalRT.computeIfAbsent(metric, k -> new AtomicLong());
-            total.addAndGet(responseTime);
-
-            AtomicLong count = rtCount.computeIfAbsent(metric, k -> new AtomicLong());
-            count.incrementAndGet();
-
-            avgRT.computeIfAbsent(metric, k -> new AtomicLong());
-
-            publishEvent(new RTEvent(metric, responseTime));
-        }
-    }
-
-    private void publishEvent(MetricsEvent event) {
-        for (MetricsListener listener : listeners) {
-            listener.onEvent(event);
-        }
+        stats.addRT(interfaceName, methodName, group, version, responseTime);
     }
 
     @Override
@@ -164,24 +121,51 @@
     }
 
     private void collectRequests(List<MetricSample> list) {
-        totalRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_TOTAL, k.getTags(), REQUESTS, v::get)));
-        succeedRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_SUCCEED, k.getTags(), REQUESTS, v::get)));
-        failedRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_FAILED, k.getTags(), REQUESTS, v::get)));
-        processingRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_PROCESSING, k.getTags(), REQUESTS, v::get)));
+        doExecute(RequestEvent.Type.TOTAL, MetricsStatHandler::get).filter(e->!e.isEmpty())
+            .ifPresent(map-> map.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_TOTAL, k.getTags(), REQUESTS, v::get))));
+
+        doExecute(RequestEvent.Type.SUCCEED, MetricsStatHandler::get).filter(e->!e.isEmpty())
+            .ifPresent(map-> map.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_SUCCEED, k.getTags(), REQUESTS, v::get))));
+
+        doExecute(RequestEvent.Type.FAILED, MetricsStatHandler::get).filter(e->!e.isEmpty())
+            .ifPresent(map->{
+                map.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_FAILED, k.getTags(), REQUESTS, v::get)));
+            });
+
+        doExecute(RequestEvent.Type.PROCESSING, MetricsStatHandler::get).filter(e->!e.isEmpty())
+            .ifPresent(map-> map.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_PROCESSING, k.getTags(), REQUESTS, v::get))));
+
+        doExecute(RequestEvent.Type.BUSINESS_FAILED, MetricsStatHandler::get).filter(e->!e.isEmpty())
+            .ifPresent(map-> map.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUEST_BUSINESS_FAILED, k.getTags(), REQUESTS, v::get))));
     }
 
     private void collectRT(List<MetricSample> list) {
-        lastRT.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_LAST, k.getTags(), RT, v::get)));
-        minRT.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_MIN, k.getTags(), RT, v::get)));
-        maxRT.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_MAX, k.getTags(), RT, v::get)));
+        this.stats.getLastRT().forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_LAST, k.getTags(), RT, v::get)));
+        this.stats.getMinRT().forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_MIN, k.getTags(), RT, v::get)));
+        this.stats.getMaxRT().forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_MAX, k.getTags(), RT, v::get)));
 
-        totalRT.forEach((k, v) -> {
+        this.stats.getTotalRT().forEach((k, v) -> {
             list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_TOTAL, k.getTags(), RT, v::get));
 
-            AtomicLong avg = avgRT.get(k);
-            AtomicLong count = rtCount.get(k);
+            AtomicLong avg = this.stats.getAvgRT().get(k);
+            AtomicLong count = this.stats.getRtCount().get(k);
             avg.set(v.get() / count.get());
             list.add(new GaugeMetricSample(MetricsKey.METRIC_RT_AVG, k.getTags(), RT, avg::get));
         });
     }
+    private <T> Optional<T> doExecute(RequestEvent.Type requestType, Function<MetricsStatHandler,T> statExecutor) {
+        if (isCollectEnabled()) {
+            MetricsStatHandler handler = stats.getHandler(requestType);
+            T result =  statExecutor.apply(handler);
+            return Optional.ofNullable(result);
+        }
+        return Optional.empty();
+    }
+
+    private void doExecute(RequestEvent.Type requestType, Consumer<MetricsStatHandler> statExecutor) {
+        if (isCollectEnabled()) {
+            MetricsStatHandler handler = stats.getHandler(requestType);
+             statExecutor.accept(handler);
+        }
+    }
 }
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/DefaultMetricsStatHandler.java b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/DefaultMetricsStatHandler.java
new file mode 100644
index 0000000..1670b9d
--- /dev/null
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/DefaultMetricsStatHandler.java
@@ -0,0 +1,74 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.dubbo.common.metrics.collector.stat;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BiConsumer;
+
+import org.apache.dubbo.common.metrics.model.MethodMetric;
+
+
+public class DefaultMetricsStatHandler implements MetricsStatHandler {
+
+    private final String applicationName;
+    private final Map<MethodMetric, AtomicLong> counts = new ConcurrentHashMap<>();
+
+    public DefaultMetricsStatHandler(String applicationName) {
+        this.applicationName = applicationName;
+    }
+
+    @Override
+    public void increase(String interfaceName, String methodName, String group, String version) {
+        this.doIncrExecute(interfaceName,methodName,group,version);
+    }
+
+    public void decrease(String interfaceName, String methodName, String group, String version){
+        this.doDecrExecute(interfaceName,methodName,group,version);
+    }
+
+    protected void doExecute(String interfaceName, String methodName, String group, String version, BiConsumer<MethodMetric,Map<MethodMetric, AtomicLong>> execute){
+        MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
+        execute.accept(metric,counts);
+
+        this.doNotify(metric);
+    }
+
+    protected void doIncrExecute(String interfaceName, String methodName, String group, String version){
+        this.doExecute(interfaceName,methodName,group,version,(metric,counts)->{
+            AtomicLong count = counts.computeIfAbsent(metric, k -> new AtomicLong(0L));
+            count.incrementAndGet();
+
+        });
+    }
+
+    protected void doDecrExecute(String interfaceName, String methodName, String group, String version){
+        this.doExecute(interfaceName,methodName,group,version,(metric,counts)->{
+            AtomicLong count = counts.computeIfAbsent(metric, k -> new AtomicLong(0L));
+            count.decrementAndGet();
+        });
+    }
+
+    @Override
+    public Map<MethodMetric, AtomicLong> get() {
+        return counts;
+    }
+
+    public  void doNotify(MethodMetric metric){}
+
+}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java
new file mode 100644
index 0000000..d86a206
--- /dev/null
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatComposite.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dubbo.common.metrics.collector.stat;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.LongAccumulator;
+
+import org.apache.dubbo.common.metrics.collector.DefaultMetricsCollector;
+import org.apache.dubbo.common.metrics.event.MetricsEvent;
+import org.apache.dubbo.common.metrics.event.RTEvent;
+import org.apache.dubbo.common.metrics.event.RequestEvent;
+import org.apache.dubbo.common.metrics.listener.MetricsListener;
+import org.apache.dubbo.common.metrics.model.MethodMetric;
+public class MetricsStatComposite{
+
+    public Map<RequestEvent.Type, MetricsStatHandler> stats = new ConcurrentHashMap<>();
+    private final Map<MethodMetric, AtomicLong>     lastRT = new ConcurrentHashMap<>();
+    private final Map<MethodMetric, LongAccumulator> minRT  = new ConcurrentHashMap<>();
+    private final Map<MethodMetric, LongAccumulator> maxRT  = new ConcurrentHashMap<>();
+    private final Map<MethodMetric, AtomicLong> avgRT = new ConcurrentHashMap<>();
+    private final Map<MethodMetric, AtomicLong> totalRT = new ConcurrentHashMap<>();
+    private final Map<MethodMetric, AtomicLong> rtCount = new ConcurrentHashMap<>();
+    private final String applicationName;
+    private final List<MetricsListener> listeners;
+    private DefaultMetricsCollector collector;
+
+    public MetricsStatComposite(String applicationName, DefaultMetricsCollector collector){
+        this.applicationName = applicationName;
+        this.listeners = collector.getListener();
+        this.collector = collector;
+        this.init();
+    }
+
+    public MetricsStatHandler getHandler(RequestEvent.Type statType) {
+        return stats.get(statType);
+    }
+
+    public Map<MethodMetric, AtomicLong> getLastRT(){
+        return this.lastRT;
+    }
+    public Map<MethodMetric, LongAccumulator> getMinRT(){
+        return this.minRT;
+    }
+
+    public Map<MethodMetric, LongAccumulator> getMaxRT(){
+        return this.maxRT;
+    }
+    public Map<MethodMetric, AtomicLong> getAvgRT(){
+        return this.avgRT;
+    }
+    public Map<MethodMetric, AtomicLong> getTotalRT(){
+        return this.totalRT;
+    }
+    public Map<MethodMetric, AtomicLong> getRtCount(){
+        return this.rtCount;
+    }
+
+    public void addRT(String interfaceName, String methodName, String group, String version, Long responseTime) {
+        if (collector.isCollectEnabled()) {
+            MethodMetric metric = new MethodMetric(applicationName, interfaceName, methodName, group, version);
+
+            AtomicLong last = lastRT.computeIfAbsent(metric, k -> new AtomicLong());
+            last.set(responseTime);
+
+            LongAccumulator min = minRT.computeIfAbsent(metric, k -> new LongAccumulator(Long::min, Long.MAX_VALUE));
+            min.accumulate(responseTime);
+
+            LongAccumulator max = maxRT.computeIfAbsent(metric, k -> new LongAccumulator(Long::max, Long.MIN_VALUE));
+            max.accumulate(responseTime);
+
+            AtomicLong total = totalRT.computeIfAbsent(metric, k -> new AtomicLong());
+            total.addAndGet(responseTime);
+
+            AtomicLong count = rtCount.computeIfAbsent(metric, k -> new AtomicLong());
+            count.incrementAndGet();
+
+            avgRT.computeIfAbsent(metric, k -> new AtomicLong());
+
+            publishEvent(new RTEvent(metric, responseTime));
+        }
+    }
+
+    private void init() {
+        stats.put(RequestEvent.Type.TOTAL, new DefaultMetricsStatHandler(applicationName){
+            @Override
+            public void doNotify(MethodMetric metric) {
+                publishEvent(new RequestEvent(metric, RequestEvent.Type.TOTAL));
+            }
+        });
+
+        stats.put(RequestEvent.Type.SUCCEED, new DefaultMetricsStatHandler(applicationName) {
+            @Override
+            public void doNotify(MethodMetric metric) {
+                publishEvent(new RequestEvent(metric, RequestEvent.Type.SUCCEED));
+            }
+        });
+
+        stats.put(RequestEvent.Type.FAILED, new DefaultMetricsStatHandler(applicationName) {
+            @Override
+            public void doNotify(MethodMetric metric) {
+                publishEvent(new RequestEvent(metric, RequestEvent.Type.FAILED));
+            }
+        });
+
+        stats.put(RequestEvent.Type.BUSINESS_FAILED, new DefaultMetricsStatHandler(applicationName) {
+            @Override
+            public void doNotify(MethodMetric metric) {
+                publishEvent(new RequestEvent(metric, RequestEvent.Type.BUSINESS_FAILED));
+            }
+        });
+
+        stats.put(RequestEvent.Type.PROCESSING, new DefaultMetricsStatHandler(applicationName));
+    }
+
+    private void publishEvent(MetricsEvent event) {
+        for (MetricsListener listener : listeners) {
+            listener.onEvent(event);
+        }
+    }
+}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatHandler.java b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatHandler.java
new file mode 100644
index 0000000..d39c5d4
--- /dev/null
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/collector/stat/MetricsStatHandler.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dubbo.common.metrics.collector.stat;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.dubbo.common.metrics.model.MethodMetric;
+
+public interface MetricsStatHandler {
+    Map<MethodMetric, AtomicLong> get();
+    void increase(String interfaceName, String methodName, String group, String version);
+    void decrease(String interfaceName, String methodName, String group, String version);
+}
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/event/RequestEvent.java b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/event/RequestEvent.java
index b1d050b..f0a6677 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/event/RequestEvent.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/event/RequestEvent.java
@@ -40,6 +40,8 @@
         TOTAL,
         SUCCEED,
         FAILED,
+        BUSINESS_FAILED,
+
         PROCESSING
     }
 }
diff --git a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/model/MetricsKey.java b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/model/MetricsKey.java
index 07816a3..bbb1bd8 100644
--- a/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/model/MetricsKey.java
+++ b/dubbo-common/src/main/java/org/apache/dubbo/common/metrics/model/MetricsKey.java
@@ -22,10 +22,14 @@
     METRIC_REQUESTS_TOTAL("requests.total", "Total Requests"),
     METRIC_REQUESTS_SUCCEED("requests.succeed", "Succeed Requests"),
     METRIC_REQUESTS_FAILED("requests.failed", "Failed Requests"),
+    METRIC_REQUEST_BUSINESS_FAILED("requests.business.failed","Failed Business Requests"),
     METRIC_REQUESTS_PROCESSING("requests.processing", "Processing Requests"),
+
     METRIC_REQUESTS_TOTAL_AGG("requests.total.aggregate", "Aggregated Total Requests"),
     METRIC_REQUESTS_SUCCEED_AGG("requests.succeed.aggregate", "Aggregated Succeed Requests"),
     METRIC_REQUESTS_FAILED_AGG("requests.failed.aggregate", "Aggregated Failed Requests"),
+    METRIC_REQUESTS_BUSINESS_FAILED_AGG("requests.business.failed.aggregate", "Aggregated Business Failed Requests"),
+
     METRIC_QPS("qps", "Query Per Seconds"),
     METRIC_RT_LAST("rt.last", "Last Response Time"),
     METRIC_RT_MIN("rt.min", "Min Response Time"),
diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
index b473d1f..302c851 100644
--- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
+++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollector.java
@@ -54,6 +54,7 @@
     private final Map<MethodMetric, TimeWindowCounter> totalRequests = new ConcurrentHashMap<>();
     private final Map<MethodMetric, TimeWindowCounter> succeedRequests = new ConcurrentHashMap<>();
     private final Map<MethodMetric, TimeWindowCounter> failedRequests = new ConcurrentHashMap<>();
+    private final Map<MethodMetric, TimeWindowCounter> businessFailedRequests = new ConcurrentHashMap<>();
     private final Map<MethodMetric, TimeWindowCounter> qps = new ConcurrentHashMap<>();
     private final Map<MethodMetric, TimeWindowQuantile> rt = new ConcurrentHashMap<>();
 
@@ -113,6 +114,10 @@
             case FAILED:
                 counter = failedRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
                 break;
+            case BUSINESS_FAILED:
+                counter = businessFailedRequests.computeIfAbsent(metric, k -> new TimeWindowCounter(bucketNum, timeWindowSeconds));
+                break;
+
             default:
                 break;
         }
@@ -136,6 +141,7 @@
         totalRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_TOTAL_AGG, k.getTags(), REQUESTS, v::get)));
         succeedRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_SUCCEED_AGG, k.getTags(), REQUESTS, v::get)));
         failedRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_FAILED_AGG, k.getTags(), REQUESTS, v::get)));
+        businessFailedRequests.forEach((k, v) -> list.add(new GaugeMetricSample(MetricsKey.METRIC_REQUESTS_BUSINESS_FAILED_AGG, k.getTags(), REQUESTS, v::get)));
     }
 
     private void collectQPS(List<MetricSample> list) {
diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/filter/MetricsCollectExecutor.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/filter/MetricsCollectExecutor.java
new file mode 100644
index 0000000..0ae3fdf
--- /dev/null
+++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/filter/MetricsCollectExecutor.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dubbo.metrics.filter;
+
+import static org.apache.dubbo.common.constants.MetricsConstants.METRIC_FILTER_START_TIME;
+
+import java.util.function.Supplier;
+
+import org.apache.dubbo.common.metrics.collector.DefaultMetricsCollector;
+import org.apache.dubbo.rpc.Invocation;
+import org.apache.dubbo.rpc.Result;
+import org.apache.dubbo.rpc.RpcException;
+
+public class MetricsCollectExecutor {
+
+    private final DefaultMetricsCollector collector;
+    private final Invocation    invocation;
+    private String  interfaceName;
+    private String  methodName;
+    private String  group;
+    private String  version;
+
+
+    public MetricsCollectExecutor(DefaultMetricsCollector collector, Invocation invocation) {
+        init(invocation);
+
+        this.collector = collector;
+
+        this.invocation = invocation;
+    }
+
+    public void beforeExecute() {
+        collector.increaseTotalRequests(interfaceName, methodName, group, version);
+        collector.increaseProcessingRequests(interfaceName, methodName, group, version);
+        invocation.put(METRIC_FILTER_START_TIME, System.currentTimeMillis());
+    }
+
+    public void postExecute(Result result) {
+        if (result.hasException()) {
+            this.throwExecute(result.getException());
+            return;
+        }
+        collector.increaseSucceedRequests(interfaceName, methodName, group, version);
+        endExecute();
+    }
+
+    public void throwExecute(Throwable throwable){
+        if (throwable instanceof RpcException) {
+            RpcException rpcException = (RpcException)throwable;
+            if (rpcException.isBiz()) {
+                collector.businessFailedRequests(interfaceName, methodName, group, version);
+            }else{
+                collector.increaseFailedRequests(interfaceName, methodName, group, version);
+            }
+        }
+        endExecute(()-> throwable instanceof RpcException && ((RpcException) throwable).isBiz());
+    }
+
+    private void endExecute(){
+        this.endExecute(() -> true);
+    }
+
+    private void endExecute(Supplier<Boolean> rtStat){
+        if (rtStat.get()) {
+            Long endTime = System.currentTimeMillis();
+            Long beginTime = (Long) invocation.get(METRIC_FILTER_START_TIME);
+            Long rt = endTime - beginTime;
+            collector.addRT(interfaceName, methodName, group, version, rt);
+        }
+        collector.decreaseProcessingRequests(interfaceName, methodName, group, version);
+    }
+
+    private void init(Invocation invocation) {
+        String serviceUniqueName = invocation.getTargetServiceUniqueName();
+        String methodName = invocation.getMethodName();
+        String group = null;
+        String interfaceAndVersion;
+        String[] arr = serviceUniqueName.split("/");
+        if (arr.length == 2) {
+            group = arr[0];
+            interfaceAndVersion = arr[1];
+        } else {
+            interfaceAndVersion = arr[0];
+        }
+
+        String[] ivArr = interfaceAndVersion.split(":");
+        String interfaceName = ivArr[0];
+        String version = ivArr.length == 2 ? ivArr[1] : null;
+
+        this.interfaceName = interfaceName;
+        this.methodName = methodName;
+        this.group = group;
+        this.version = version;
+    }
+}
diff --git a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/filter/MetricsFilter.java b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/filter/MetricsFilter.java
index 031f8ef..4e5cc13 100644
--- a/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/filter/MetricsFilter.java
+++ b/dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/filter/MetricsFilter.java
@@ -14,11 +14,15 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.dubbo.metrics.filter;
 
+import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
+
+import java.util.function.Consumer;
+
 import org.apache.dubbo.common.extension.Activate;
 import org.apache.dubbo.common.metrics.collector.DefaultMetricsCollector;
+import org.apache.dubbo.rpc.BaseFilter;
 import org.apache.dubbo.rpc.Filter;
 import org.apache.dubbo.rpc.Invocation;
 import org.apache.dubbo.rpc.Invoker;
@@ -27,19 +31,17 @@
 import org.apache.dubbo.rpc.model.ApplicationModel;
 import org.apache.dubbo.rpc.model.ScopeModelAware;
 
-import static org.apache.dubbo.common.constants.CommonConstants.PROVIDER;
-
 @Activate(group = PROVIDER, order = -1)
-public class MetricsFilter implements Filter, ScopeModelAware {
+public class MetricsFilter implements Filter, BaseFilter.Listener, ScopeModelAware {
 
     private DefaultMetricsCollector collector = null;
 
     private ApplicationModel applicationModel;
 
+
     @Override
     public void setApplicationModel(ApplicationModel applicationModel) {
         this.applicationModel = applicationModel;
-
         collector = applicationModel.getBeanFactory().getBean(DefaultMetricsCollector.class);
     }
 
@@ -48,38 +50,23 @@
         if (collector == null || !collector.isCollectEnabled()) {
             return invoker.invoke(invocation);
         }
+        collect(invocation, MetricsCollectExecutor::beforeExecute);
 
-        String serviceUniqueName = invocation.getTargetServiceUniqueName();
-        String methodName = invocation.getMethodName();
-        String group = null;
-        String interfaceAndVersion;
-        String[] arr = serviceUniqueName.split("/");
-        if (arr.length == 2) {
-            group = arr[0];
-            interfaceAndVersion = arr[1];
-        } else {
-            interfaceAndVersion = arr[0];
-        }
+        return invoker.invoke(invocation);
+    }
 
-        String[] ivArr = interfaceAndVersion.split(":");
-        String interfaceName = ivArr[0];
-        String version = ivArr.length == 2 ? ivArr[1] : null;
-        collector.increaseTotalRequests(interfaceName, methodName, group, version);
-        collector.increaseProcessingRequests(interfaceName, methodName, group, version);
+    @Override
+    public void onResponse(Result result, Invoker<?> invoker, Invocation invocation) {
+        collect(invocation, collector->collector.postExecute(result));
+    }
 
-        Long startTime = System.currentTimeMillis();
-        try {
-            Result invoke = invoker.invoke(invocation);
-            collector.increaseSucceedRequests(interfaceName, methodName, group, version);
-            return invoke;
-        } catch (RpcException e) {
-            collector.increaseFailedRequests(interfaceName, methodName, group, version);
-            throw e;
-        } finally {
-            Long endTime = System.currentTimeMillis();
-            Long rt = endTime - startTime;
-            collector.addRT(interfaceName, methodName, group, version, rt);
-            collector.decreaseProcessingRequests(interfaceName, methodName, group, version);
-        }
+    @Override
+    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
+        collect(invocation,collector-> collector.throwExecute(t));
+    }
+
+    private void collect(Invocation invocation, Consumer<MetricsCollectExecutor> execute) {
+        MetricsCollectExecutor collectorExecutor = new MetricsCollectExecutor(collector, invocation);
+        execute.accept(collectorExecutor);
     }
 }
diff --git a/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollectorTest.java b/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollectorTest.java
index 308e238..0b3317a 100644
--- a/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollectorTest.java
+++ b/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/collector/AggregateMetricsCollectorTest.java
@@ -18,6 +18,7 @@
 package org.apache.dubbo.metrics.collector;
 
 import org.apache.dubbo.common.metrics.collector.DefaultMetricsCollector;
+import org.apache.dubbo.common.metrics.model.MetricsKey;
 import org.apache.dubbo.common.metrics.model.sample.GaugeMetricSample;
 import org.apache.dubbo.common.metrics.model.sample.MetricSample;
 import org.apache.dubbo.config.ApplicationConfig;
@@ -81,6 +82,7 @@
         defaultCollector.increaseTotalRequests(interfaceName, methodName, group, version);
         defaultCollector.increaseSucceedRequests(interfaceName, methodName, group, version);
         defaultCollector.increaseFailedRequests(interfaceName, methodName, group, version);
+        defaultCollector.businessFailedRequests(interfaceName,methodName,group,version);
 
         List<MetricSample> samples = collector.collect();
         for (MetricSample sample : samples) {
@@ -101,6 +103,8 @@
         Assertions.assertEquals(sampleMap.get("requests.total.aggregate"), 1L);
         Assertions.assertEquals(sampleMap.get("requests.succeed.aggregate"), 1L);
         Assertions.assertEquals(sampleMap.get("requests.failed.aggregate"), 1L);
+        Assertions.assertEquals(sampleMap.get(MetricsKey.METRIC_REQUESTS_BUSINESS_FAILED_AGG.getName()), 1L);
+
         Assertions.assertTrue(sampleMap.containsKey("qps"));
     }
 
diff --git a/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/filter/MetricsFilterTest.java b/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/filter/MetricsFilterTest.java
index 2e86a83..dbe6806 100644
--- a/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/filter/MetricsFilterTest.java
+++ b/dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/filter/MetricsFilterTest.java
@@ -17,11 +17,25 @@
 
 package org.apache.dubbo.metrics.filter;
 
+import static org.apache.dubbo.common.constants.MetricsConstants.TAG_GROUP_KEY;
+import static org.apache.dubbo.common.constants.MetricsConstants.TAG_INTERFACE_KEY;
+import static org.apache.dubbo.common.constants.MetricsConstants.TAG_METHOD_KEY;
+import static org.apache.dubbo.common.constants.MetricsConstants.TAG_VERSION_KEY;
+import static org.mockito.BDDMockito.given;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
 import org.apache.dubbo.common.metrics.collector.DefaultMetricsCollector;
+import org.apache.dubbo.common.metrics.model.MetricsKey;
 import org.apache.dubbo.common.metrics.model.sample.MetricSample;
 import org.apache.dubbo.config.ApplicationConfig;
 import org.apache.dubbo.rpc.AppResponse;
 import org.apache.dubbo.rpc.Invoker;
+import org.apache.dubbo.rpc.Result;
 import org.apache.dubbo.rpc.RpcException;
 import org.apache.dubbo.rpc.RpcInvocation;
 import org.apache.dubbo.rpc.model.ApplicationModel;
@@ -30,15 +44,6 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import java.util.List;
-import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
-
-import static org.apache.dubbo.common.constants.MetricsConstants.*;
-import static org.mockito.BDDMockito.given;
-import static org.mockito.Mockito.mock;
-
 public class MetricsFilterTest {
 
     private ApplicationModel applicationModel;
@@ -89,8 +94,9 @@
 
         try {
             filter.invoke(invoker, invocation);
-        } catch (Exception ignore) {
-
+        } catch (Exception e) {
+            Assertions.assertTrue(e instanceof RpcException);
+            filter.onError(e, invoker, invocation);
         }
 
         Map<String, MetricSample> metricsMap = getMetricsMap();
@@ -106,13 +112,45 @@
         Assertions.assertEquals(tags.get(TAG_VERSION_KEY), VERSION);
     }
 
+
+    @Test
+    public void testBusinessFailedRequests() {
+        collector.setCollectEnabled(true);
+
+        given(invoker.invoke(invocation)).willThrow(new RpcException(RpcException.BIZ_EXCEPTION));
+        initParam();
+
+        try {
+            filter.invoke(invoker, invocation);
+        } catch (Exception e) {
+            Assertions.assertTrue(e instanceof RpcException);
+            filter.onError(e, invoker, invocation);
+        }
+
+        Map<String, MetricSample> metricsMap = getMetricsMap();
+        Assertions.assertTrue(metricsMap.containsKey(MetricsKey.METRIC_REQUEST_BUSINESS_FAILED.getName()));
+        Assertions.assertFalse(metricsMap.containsKey("requests.succeed"));
+
+        MetricSample sample = metricsMap.get(MetricsKey.METRIC_REQUEST_BUSINESS_FAILED.getName());
+
+        Map<String, String> tags = sample.getTags();
+
+        Assertions.assertEquals(tags.get(TAG_INTERFACE_KEY), INTERFACE_NAME);
+        Assertions.assertEquals(tags.get(TAG_METHOD_KEY), METHOD_NAME);
+        Assertions.assertEquals(tags.get(TAG_GROUP_KEY), GROUP);
+        Assertions.assertEquals(tags.get(TAG_VERSION_KEY), VERSION);
+    }
+
     @Test
     public void testSucceedRequests() {
         collector.setCollectEnabled(true);
         given(invoker.invoke(invocation)).willReturn(new AppResponse("success"));
         initParam();
 
-        filter.invoke(invoker, invocation);
+        Result result = filter.invoke(invoker, invocation);
+
+        filter.onResponse(result, invoker, invocation);
+
         Map<String, MetricSample> metricsMap = getMetricsMap();
         Assertions.assertFalse(metricsMap.containsKey("requests.failed"));
         Assertions.assertTrue(metricsMap.containsKey("requests.succeed"));
@@ -134,7 +172,10 @@
         invocation.setMethodName(METHOD_NAME);
         invocation.setParameterTypes(new Class[]{String.class});
 
-        filter.invoke(invoker, invocation);
+        Result result = filter.invoke(invoker, invocation);
+
+        filter.onResponse(result, invoker, invocation);
+
         Map<String, MetricSample> metricsMap = getMetricsMap();
 
         MetricSample sample = metricsMap.get("requests.succeed");
@@ -154,7 +195,10 @@
         invocation.setMethodName(METHOD_NAME);
         invocation.setParameterTypes(new Class[]{String.class});
 
-        filter.invoke(invoker, invocation);
+        Result result = filter.invoke(invoker, invocation);
+
+        filter.onResponse(result, invoker, invocation);
+
         Map<String, MetricSample> metricsMap = getMetricsMap();
 
         MetricSample sample = metricsMap.get("requests.succeed");
@@ -174,7 +218,10 @@
         invocation.setMethodName(METHOD_NAME);
         invocation.setParameterTypes(new Class[]{String.class});
 
-        filter.invoke(invoker, invocation);
+        Result result = filter.invoke(invoker, invocation);
+
+        filter.onResponse(result, invoker, invocation);
+
         Map<String, MetricSample> metricsMap = getMetricsMap();
 
         MetricSample sample = metricsMap.get("requests.succeed");