1 /* 2 * Licensed to the Apache Software Foundation (ASF) under one or more 3 * contributor license agreements. See the NOTICE file distributed with 4 * this work for additional information regarding copyright ownership. 5 * The ASF licenses this file to You under the Apache License, Version 2.0 6 * (the "License"); you may not use this file except in compliance with 7 * the License. You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 */ 17 package com.alibaba.dubbo.rpc.cluster.support; 18 19 import com.alibaba.dubbo.common.Constants; 20 import com.alibaba.dubbo.common.Version; 21 import com.alibaba.dubbo.common.logger.Logger; 22 import com.alibaba.dubbo.common.logger.LoggerFactory; 23 import com.alibaba.dubbo.common.utils.NetUtils; 24 import com.alibaba.dubbo.rpc.Invocation; 25 import com.alibaba.dubbo.rpc.Invoker; 26 import com.alibaba.dubbo.rpc.Result; 27 import com.alibaba.dubbo.rpc.RpcContext; 28 import com.alibaba.dubbo.rpc.RpcException; 29 import com.alibaba.dubbo.rpc.cluster.Directory; 30 import com.alibaba.dubbo.rpc.cluster.LoadBalance; 31 32 import java.util.ArrayList; 33 import java.util.HashSet; 34 import java.util.List; 35 import java.util.Set; 36 37 /** 38 * When invoke fails, log the initial error and retry other invokers (retry n times, which means at most n different invokers will be invoked) 39 * Note that retry causes latency. 40 * <p> 41 * <a href="http://en.wikipedia.org/wiki/Failover">Failover</a> 42 * 43 */ 44 public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> { 45 46 private static final Logger logger = LoggerFactory.getLogger(FailoverClusterInvoker.class); 47 48 public FailoverClusterInvoker(Directory<T> directory) { 49 super(directory); 50 } 51 52 @Override 53 @SuppressWarnings({"unchecked", "rawtypes"}) 54 public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 55 List<Invoker<T>> copyinvokers = invokers; 56 checkInvokers(copyinvokers, invocation); 57 int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) + 1; 58 if (len <= 0) { 59 len = 1; 60 } 61 // retry loop. 62 RpcException le = null; // last exception. 63 List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers. 64 Set<String> providers = new HashSet<String>(len); 65 for (int i = 0; i < len; i++) { 66 //Reselect before retry to avoid a change of candidate `invokers`. 67 //NOTE: if `invokers` changed, then `invoked` also lose accuracy. 68 if (i > 0) { 69 checkWhetherDestroyed(); 70 copyinvokers = list(invocation); 71 // check again 72 checkInvokers(copyinvokers, invocation); 73 } 74 Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked); 75 invoked.add(invoker); 76 RpcContext.getContext().setInvokers((List) invoked); 77 try { 78 Result result = invoker.invoke(invocation); 79 if (le != null && logger.isWarnEnabled()) { 80 logger.warn("Although retry the method " + invocation.getMethodName() 81 + " in the service " + getInterface().getName() 82 + " was successful by the provider " + invoker.getUrl().getAddress() 83 + ", but there have been failed providers " + providers 84 + " (" + providers.size() + "/" + copyinvokers.size() 85 + ") from the registry " + directory.getUrl().getAddress() 86 + " on the consumer " + NetUtils.getLocalHost() 87 + " using the dubbo version " + Version.getVersion() + ". Last error is: " 88 + le.getMessage(), le); 89 } 90 return result; 91 } catch (RpcException e) { 92 if (e.isBiz()) { // biz exception. 93 throw e; 94 } 95 le = e; 96 } catch (Throwable e) { 97 le = new RpcException(e.getMessage(), e); 98 } finally { 99 providers.add(invoker.getUrl().getAddress()); 100 } 101 } 102 throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method " 103 + invocation.getMethodName() + " in the service " + getInterface().getName() 104 + ". Tried " + len + " times of the providers " + providers 105 + " (" + providers.size() + "/" + copyinvokers.size() 106 + ") from the registry " + directory.getUrl().getAddress() 107 + " on the consumer " + NetUtils.getLocalHost() + " using the dubbo version " 108 + Version.getVersion() + ". Last error is: " 109 + (le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le); 110 } 111 112 }
1 /* 2 * Licensed to the Apache Software Foundation (ASF) under one or more 3 * contributor license agreements. See the NOTICE file distributed with 4 * this work for additional information regarding copyright ownership. 5 * The ASF licenses this file to You under the Apache License, Version 2.0 6 * (the "License"); you may not use this file except in compliance with 7 * the License. You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 */ 17 package com.alibaba.dubbo.rpc.cluster.support; 18 19 import com.alibaba.dubbo.common.Version; 20 import com.alibaba.dubbo.common.utils.NetUtils; 21 import com.alibaba.dubbo.rpc.Invocation; 22 import com.alibaba.dubbo.rpc.Invoker; 23 import com.alibaba.dubbo.rpc.Result; 24 import com.alibaba.dubbo.rpc.RpcException; 25 import com.alibaba.dubbo.rpc.cluster.Directory; 26 import com.alibaba.dubbo.rpc.cluster.LoadBalance; 27 28 import java.util.List; 29 30 /** 31 * Execute exactly once, which means this policy will throw an exception immediately in case of an invocation error. 32 * Usually used for non-idempotent write operations 33 * 34 * <a href="http://en.wikipedia.org/wiki/Fail-fast">Fail-fast</a> 35 * 36 */ 37 public class FailfastClusterInvoker<T> extends AbstractClusterInvoker<T> { 38 39 public FailfastClusterInvoker(Directory<T> directory) { 40 super(directory); 41 } 42 43 @Override 44 public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 45 checkInvokers(invokers, invocation); 46 Invoker<T> invoker = select(loadbalance, invocation, invokers, null); 47 try { 48 return invoker.invoke(invocation); 49 } catch (Throwable e) { 50 if (e instanceof RpcException && ((RpcException) e).isBiz()) { // biz exception. 51 throw (RpcException) e; 52 } 53 throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failfast invoke providers " + invoker.getUrl() + " " + loadbalance.getClass().getSimpleName() + " select from all providers " + invokers + " for service " + getInterface().getName() + " method " + invocation.getMethodName() + " on consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e); 54 } 55 } 56 }
1 /* 2 * Licensed to the Apache Software Foundation (ASF) under one or more 3 * contributor license agreements. See the NOTICE file distributed with 4 * this work for additional information regarding copyright ownership. 5 * The ASF licenses this file to You under the Apache License, Version 2.0 6 * (the "License"); you may not use this file except in compliance with 7 * the License. You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 */ 17 package com.alibaba.dubbo.rpc.cluster.support; 18 19 import com.alibaba.dubbo.common.logger.Logger; 20 import com.alibaba.dubbo.common.logger.LoggerFactory; 21 import com.alibaba.dubbo.rpc.Invocation; 22 import com.alibaba.dubbo.rpc.Invoker; 23 import com.alibaba.dubbo.rpc.Result; 24 import com.alibaba.dubbo.rpc.RpcException; 25 import com.alibaba.dubbo.rpc.RpcResult; 26 import com.alibaba.dubbo.rpc.cluster.Directory; 27 import com.alibaba.dubbo.rpc.cluster.LoadBalance; 28 29 import java.util.List; 30 31 /** 32 * When invoke fails, log the error message and ignore this error by returning an empty RpcResult. 33 * Usually used to write audit logs and other operations 34 * 35 * <a href="http://en.wikipedia.org/wiki/Fail-safe">Fail-safe</a> 36 * 37 */ 38 public class FailsafeClusterInvoker<T> extends AbstractClusterInvoker<T> { 39 private static final Logger logger = LoggerFactory.getLogger(FailsafeClusterInvoker.class); 40 41 public FailsafeClusterInvoker(Directory<T> directory) { 42 super(directory); 43 } 44 45 @Override 46 public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 47 try { 48 checkInvokers(invokers, invocation); 49 Invoker<T> invoker = select(loadbalance, invocation, invokers, null); 50 return invoker.invoke(invocation); 51 } catch (Throwable e) { 52 logger.error("Failsafe ignore exception: " + e.getMessage(), e); 53 return new RpcResult(); // ignore 54 } 55 } 56 }
1 /* 2 * Licensed to the Apache Software Foundation (ASF) under one or more 3 * contributor license agreements. See the NOTICE file distributed with 4 * this work for additional information regarding copyright ownership. 5 * The ASF licenses this file to You under the Apache License, Version 2.0 6 * (the "License"); you may not use this file except in compliance with 7 * the License. You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 */ 17 package com.alibaba.dubbo.rpc.cluster.support; 18 19 import com.alibaba.dubbo.common.logger.Logger; 20 import com.alibaba.dubbo.common.logger.LoggerFactory; 21 import com.alibaba.dubbo.common.utils.NamedThreadFactory; 22 import com.alibaba.dubbo.rpc.Invocation; 23 import com.alibaba.dubbo.rpc.Invoker; 24 import com.alibaba.dubbo.rpc.Result; 25 import com.alibaba.dubbo.rpc.RpcException; 26 import com.alibaba.dubbo.rpc.RpcResult; 27 import com.alibaba.dubbo.rpc.cluster.Directory; 28 import com.alibaba.dubbo.rpc.cluster.LoadBalance; 29 30 import java.util.HashMap; 31 import java.util.List; 32 import java.util.Map; 33 import java.util.concurrent.ConcurrentHashMap; 34 import java.util.concurrent.ConcurrentMap; 35 import java.util.concurrent.Executors; 36 import java.util.concurrent.ScheduledExecutorService; 37 import java.util.concurrent.ScheduledFuture; 38 import java.util.concurrent.TimeUnit; 39 40 /** 41 * When fails, record failure requests and schedule for retry on a regular interval. 42 * Especially useful for services of notification. 43 * 44 * <a href="http://en.wikipedia.org/wiki/Failback">Failback</a> 45 * 46 */ 47 public class FailbackClusterInvoker<T> extends AbstractClusterInvoker<T> { 48 49 private static final Logger logger = LoggerFactory.getLogger(FailbackClusterInvoker.class); 50 51 private static final long RETRY_FAILED_PERIOD = 5 * 1000; 52 53 private final ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2, new NamedThreadFactory("failback-cluster-timer", true)); 54 private final ConcurrentMap<Invocation, AbstractClusterInvoker<?>> failed = new ConcurrentHashMap<Invocation, AbstractClusterInvoker<?>>(); 55 private volatile ScheduledFuture<?> retryFuture; 56 57 public FailbackClusterInvoker(Directory<T> directory) { 58 super(directory); 59 } 60 61 private void addFailed(Invocation invocation, AbstractClusterInvoker<?> router) { 62 if (retryFuture == null) { 63 synchronized (this) { 64 if (retryFuture == null) { 65 retryFuture = scheduledExecutorService.scheduleWithFixedDelay(new Runnable() { 66 67 @Override 68 public void run() { 69 // collect retry statistics 70 try { 71 retryFailed(); 72 } catch (Throwable t) { // Defensive fault tolerance 73 logger.error("Unexpected error occur at collect statistic", t); 74 } 75 } 76 }, RETRY_FAILED_PERIOD, RETRY_FAILED_PERIOD, TimeUnit.MILLISECONDS); 77 } 78 } 79 } 80 failed.put(invocation, router); 81 } 82 83 void retryFailed() { 84 if (failed.size() == 0) { 85 return; 86 } 87 for (Map.Entry<Invocation, AbstractClusterInvoker<?>> entry : new HashMap<Invocation, AbstractClusterInvoker<?>>( 88 failed).entrySet()) { 89 Invocation invocation = entry.getKey(); 90 Invoker<?> invoker = entry.getValue(); 91 try { 92 invoker.invoke(invocation); 93 failed.remove(invocation); 94 } catch (Throwable e) { 95 logger.error("Failed retry to invoke method " + invocation.getMethodName() + ", waiting again.", e); 96 } 97 } 98 } 99 100 @Override 101 protected Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 102 try { 103 checkInvokers(invokers, invocation); 104 Invoker<T> invoker = select(loadbalance, invocation, invokers, null); 105 return invoker.invoke(invocation); 106 } catch (Throwable e) { 107 logger.error("Failback to invoke method " + invocation.getMethodName() + ", wait for retry in background. Ignored exception: " 108 + e.getMessage() + ", ", e); 109 addFailed(invocation, this); 110 return new RpcResult(); // ignore 111 } 112 } 113 114 }
1 /* 2 * Licensed to the Apache Software Foundation (ASF) under one or more 3 * contributor license agreements. See the NOTICE file distributed with 4 * this work for additional information regarding copyright ownership. 5 * The ASF licenses this file to You under the Apache License, Version 2.0 6 * (the "License"); you may not use this file except in compliance with 7 * the License. You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 */ 17 package com.alibaba.dubbo.rpc.cluster.support; 18 19 import com.alibaba.dubbo.common.Constants; 20 import com.alibaba.dubbo.common.utils.NamedThreadFactory; 21 import com.alibaba.dubbo.rpc.Invocation; 22 import com.alibaba.dubbo.rpc.Invoker; 23 import com.alibaba.dubbo.rpc.Result; 24 import com.alibaba.dubbo.rpc.RpcContext; 25 import com.alibaba.dubbo.rpc.RpcException; 26 import com.alibaba.dubbo.rpc.cluster.Directory; 27 import com.alibaba.dubbo.rpc.cluster.LoadBalance; 28 29 import java.util.ArrayList; 30 import java.util.List; 31 import java.util.concurrent.BlockingQueue; 32 import java.util.concurrent.ExecutorService; 33 import java.util.concurrent.Executors; 34 import java.util.concurrent.LinkedBlockingQueue; 35 import java.util.concurrent.TimeUnit; 36 import java.util.concurrent.atomic.AtomicInteger; 37 38 /** 39 * Invoke a specific number of invokers concurrently, usually used for demanding real-time operations, but need to waste more service resources. 40 * 41 * <a href="http://en.wikipedia.org/wiki/Fork_(topology)">Fork</a> 42 * 43 */ 44 public class ForkingClusterInvoker<T> extends AbstractClusterInvoker<T> { 45 46 private final ExecutorService executor = Executors.newCachedThreadPool(new NamedThreadFactory("forking-cluster-timer", true)); 47 48 public ForkingClusterInvoker(Directory<T> directory) { 49 super(directory); 50 } 51 52 @Override 53 @SuppressWarnings({"unchecked", "rawtypes"}) 54 public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 55 checkInvokers(invokers, invocation); 56 final List<Invoker<T>> selected; 57 final int forks = getUrl().getParameter(Constants.FORKS_KEY, Constants.DEFAULT_FORKS); 58 final int timeout = getUrl().getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT); 59 if (forks <= 0 || forks >= invokers.size()) { 60 selected = invokers; 61 } else { 62 selected = new ArrayList<Invoker<T>>(); 63 for (int i = 0; i < forks; i++) { 64 // TODO. Add some comment here, refer chinese version for more details. 65 Invoker<T> invoker = select(loadbalance, invocation, invokers, selected); 66 if (!selected.contains(invoker)) {//Avoid add the same invoker several times. 67 selected.add(invoker); 68 } 69 } 70 } 71 RpcContext.getContext().setInvokers((List) selected); 72 final AtomicInteger count = new AtomicInteger(); 73 final BlockingQueue<Object> ref = new LinkedBlockingQueue<Object>(); 74 for (final Invoker<T> invoker : selected) { 75 executor.execute(new Runnable() { 76 @Override 77 public void run() { 78 try { 79 Result result = invoker.invoke(invocation); 80 ref.offer(result); 81 } catch (Throwable e) { 82 int value = count.incrementAndGet(); 83 if (value >= selected.size()) { 84 ref.offer(e); 85 } 86 } 87 } 88 }); 89 } 90 try { 91 Object ret = ref.poll(timeout, TimeUnit.MILLISECONDS); 92 if (ret instanceof Throwable) { 93 Throwable e = (Throwable) ret; 94 throw new RpcException(e instanceof RpcException ? ((RpcException) e).getCode() : 0, "Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e.getCause() != null ? e.getCause() : e); 95 } 96 return (Result) ret; 97 } catch (InterruptedException e) { 98 throw new RpcException("Failed to forking invoke provider " + selected + ", but no luck to perform the invocation. Last error is: " + e.getMessage(), e); 99 } 100 } 101 }
1 /* 2 * Licensed to the Apache Software Foundation (ASF) under one or more 3 * contributor license agreements. See the NOTICE file distributed with 4 * this work for additional information regarding copyright ownership. 5 * The ASF licenses this file to You under the Apache License, Version 2.0 6 * (the "License"); you may not use this file except in compliance with 7 * the License. You may obtain a copy of the License at 8 * 9 * http://www.apache.org/licenses/LICENSE-2.0 10 * 11 * Unless required by applicable law or agreed to in writing, software 12 * distributed under the License is distributed on an "AS IS" BASIS, 13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 * See the License for the specific language governing permissions and 15 * limitations under the License. 16 */ 17 package com.alibaba.dubbo.rpc.cluster.support; 18 19 import com.alibaba.dubbo.common.logger.Logger; 20 import com.alibaba.dubbo.common.logger.LoggerFactory; 21 import com.alibaba.dubbo.rpc.Invocation; 22 import com.alibaba.dubbo.rpc.Invoker; 23 import com.alibaba.dubbo.rpc.Result; 24 import com.alibaba.dubbo.rpc.RpcContext; 25 import com.alibaba.dubbo.rpc.RpcException; 26 import com.alibaba.dubbo.rpc.cluster.Directory; 27 import com.alibaba.dubbo.rpc.cluster.LoadBalance; 28 29 import java.util.List; 30 31 /** 32 * BroadcastClusterInvoker 33 * 34 */ 35 public class BroadcastClusterInvoker<T> extends AbstractClusterInvoker<T> { 36 37 private static final Logger logger = LoggerFactory.getLogger(BroadcastClusterInvoker.class); 38 39 public BroadcastClusterInvoker(Directory<T> directory) { 40 super(directory); 41 } 42 43 @Override 44 @SuppressWarnings({"unchecked", "rawtypes"}) 45 public Result doInvoke(final Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException { 46 checkInvokers(invokers, invocation); 47 RpcContext.getContext().setInvokers((List) invokers); 48 RpcException exception = null; 49 Result result = null; 50 for (Invoker<T> invoker : invokers) { 51 try { 52 result = invoker.invoke(invocation); 53 } catch (RpcException e) { 54 exception = e; 55 logger.warn(e.getMessage(), e); 56 } catch (Throwable e) { 57 exception = new RpcException(e.getMessage(), e); 58 logger.warn(e.getMessage(), e); 59 } 60 } 61 if (exception != null) { 62 throw exception; 63 } 64 return result; 65 } 66 67 }