github新增倉庫 "dubbo-read"(點此查看),集合全部《Dubbo原理和源碼解析》系列文章,後續將繼續補充該系列,同時將針對Dubbo所作的功能擴展也進行分享。不按期更新,歡迎Follow。
本文是針對 Dubbo 協議調用的調優指導,詳細說明經常使用調優參數的做用域及源碼。java
Dubbo調用模型git
經常使用性能調優參數github
參數名 | 做用範圍 | 默認值 | 說明 | 備註 |
threads | provider | 200 | 業務處理線程池大小 | |
iothreads | provider | CPU+1 | io線程池大小 | |
queues | provider | 0 | 線程池隊列大小,當線程池滿時,排隊等待執行的隊列大小,bootstrap 建議不要設置,當線程程池時應當即失敗,併發 重試其它服務提供機器,而不是排隊,除非有特殊需求ide |
|
connections | consumer | 0 | 對每一個提供者的最大鏈接數,性能 rmi、http、hessian等短鏈接協議表示限制鏈接數,this Dubbo等長鏈接協表示創建的長鏈接個數url |
Dubbo協議默認共享一個長鏈接 |
actives | consumer | 0 | 每服務消費者每服務每方法最大併發調用數 | 0表示不限制 |
acceptes | provider | 0 | 服務提供方最大可接受鏈接數 | 0表示不限制 |
executes | provider | 0 | 服務提供者每服務每方法最大可並行執行請求數 | 0表示不限制 |
源碼及原理分析spa
>> threads
FixedThreadPool.java
public Executor getExecutor(URL url) { String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME); int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS); int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES); return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, queues == 0 ? new SynchronousQueue<Runnable>() : (queues < 0 ? new LinkedBlockingQueue<Runnable>() : new LinkedBlockingQueue<Runnable>(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); }
LimitedThreadPool.java
public Executor getExecutor(URL url) { String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME); int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS); int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS); int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES); return new ThreadPoolExecutor(cores, threads, Long.MAX_VALUE, TimeUnit.MILLISECONDS, queues == 0 ? new SynchronousQueue<Runnable>() : (queues < 0 ? new LinkedBlockingQueue<Runnable>() : new LinkedBlockingQueue<Runnable>(queues)), new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url)); }
其中,Constants.DEFAULT_QUEUES = 200。threads 參數配置的是業務處理線程池的最大(或核心)線程數。
>> iothreads
NettyServer.java
@Override protected void doOpen() throws Throwable { NettyHelper.setNettyLoggerFactory(); ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true)); ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true)); ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS)); bootstrap = new ServerBootstrap(channelFactory); final NettyHandler nettyHandler = new NettyHandler(getUrl(), this); channels = nettyHandler.getChannels(); bootstrap.setPipelineFactory(new ChannelPipelineFactory() { public ChannelPipeline getPipeline() { NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this); ChannelPipeline pipeline = Channels.pipeline(); pipeline.addLast("decoder", adapter.getDecoder()); pipeline.addLast("encoder", adapter.getEncoder()); pipeline.addLast("handler", nettyHandler); return pipeline; } }); // bind channel = bootstrap.bind(getBindAddress()); }
>> queues
分別在 FixedThreadPool.java、LimitedThreadPool.java 和 CachedThreadPool.java 中使用,代碼詳情見 3.2章節。 由代碼可見,默認值爲 0,表示使用同步阻塞隊列;若是 queues 設置爲小於 0 的值,則使用容量爲 Integer.MAX_VALUE 的阻塞鏈表隊列;若是爲其餘值,則使用指定大小的阻塞鏈表隊列。
>> connections
DubboProtocol.java
private ExchangeClient[] getClients(URL url){ //是否共享鏈接 boolean service_share_connect = false; int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0); //若是connections不配置,則共享鏈接,不然每服務每鏈接 if (connections == 0){ service_share_connect = true; connections = 1; } ExchangeClient[] clients = new ExchangeClient[connections]; for (int i = 0; i < clients.length; i++) { if (service_share_connect){ clients[i] = getSharedClient(url); } else { clients[i] = initClient(url); } } return clients; }
DubboInvoker.java
@Override protected Result doInvoke(final Invocation invocation) throws Throwable { RpcInvocation inv = (RpcInvocation) invocation; final String methodName = RpcUtils.getMethodName(invocation); inv.setAttachment(Constants.PATH_KEY, getUrl().getPath()); inv.setAttachment(Constants.VERSION_KEY, version); ExchangeClient currentClient; if (clients.length == 1) { currentClient = clients[0]; } else { currentClient = clients[index.getAndIncrement() % clients.length]; } try { boolean isAsync = RpcUtils.isAsync(getUrl(), invocation); boolean isOneway = RpcUtils.isOneway(getUrl(), invocation); int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT); if (isOneway) { boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false); currentClient.send(inv, isSent); RpcContext.getContext().setFuture(null); return new RpcResult(); } else if (isAsync) { ResponseFuture future = currentClient.request(inv, timeout) ; RpcContext.getContext().setFuture(new FutureAdapter<Object>(future)); return new RpcResult(); } else { RpcContext.getContext().setFuture(null); return (Result) currentClient.request(inv, timeout).get(); } } catch (TimeoutException e) { throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } catch (RemotingException e) { throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e); } }
以上可見,默認值爲0,表示針對每一個 Provider,全部客戶端共享一個長鏈接;不然,創建指定數量的長鏈接。在調用時,若是有多個長鏈接,則使用輪詢方式得到一個長鏈接。
>> actives
ActiveLimitFilter.java
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { URL url = invoker.getUrl(); String methodName = invocation.getMethodName(); int max = invoker.getUrl().getMethodParameter(methodName, Constants.ACTIVES_KEY, 0); RpcStatus count = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()); if (max > 0) { long timeout = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, 0); long start = System.currentTimeMillis(); long remain = timeout; int active = count.getActive(); if (active >= max) { synchronized (count) { while ((active = count.getActive()) >= max) { try { count.wait(remain); } catch (InterruptedException e) { } long elapsed = System.currentTimeMillis() - start; remain = timeout - elapsed; if (remain <= 0) { throw new RpcException("Waiting concurrent invoke timeout in client-side for service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", elapsed: " + elapsed + ", timeout: " + timeout + ". concurrent invokes: " + active + ". max concurrent invoke limit: " + max); } } } } } try { long begin = System.currentTimeMillis(); RpcStatus.beginCount(url, methodName); try { Result result = invoker.invoke(invocation); RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true); return result; } catch (RuntimeException t) { RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false); throw t; } } finally { if(max>0){ synchronized (count) { count.notify(); } } } }
Consumer 調用時,統計服務和方法維度的調用狀況,若是併發數超過設置的最大值,則阻塞當前線程,直到前面有請求處理完成。
>> accepts
AbstractServer.java
@Override public void connected(Channel ch) throws RemotingException { Collection<Channel> channels = getChannels(); if (accepts > 0 && channels.size() > accepts) { logger.error("Close channel " + ch + ", cause: The server " + ch.getLocalAddress() + " connections greater than max config " + accepts); ch.close(); return; } super.connected(ch); }
當鏈接數大於最大值時,關閉當前鏈接。
>> executes
ExecuteLimitFilter.jvava
public Result invokeOrg(Invoker<?> invoker, Invocation invocation) throws RpcException { URL url = invoker.getUrl(); String methodName = invocation.getMethodName(); int max = url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0); if (max > 0) { RpcStatus count = RpcStatus.getStatus(url, invocation.getMethodName()); if (count.getActive() >= max) { throw new RpcException("Failed to invoke method " + invocation.getMethodName() + " in provider " + url + ", cause: The service using threads greater than <dubbo:service executes=\"" + max + "\" /> limited."); } } long begin = System.currentTimeMillis(); boolean isException = false; RpcStatus.beginCount(url, methodName); try { Result result = invoker.invoke(invocation); return result; } catch (Throwable t) { isException = true; if(t instanceof RuntimeException) { throw (RuntimeException) t; } else { throw new RpcException("unexpected exception when ExecuteLimitFilter", t); } } finally { RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isException); } }
Provider處理請求時,統計方法維度的調用狀況,若是併發數超過設置的最大值,則阻直接拋出異常。