本篇文章為大家展示了dubbo中ExecutionDispatcher的作用是什么,內(nèi)容簡明扼要并且容易理解,絕對能使你眼前一亮,通過這篇文章的詳細(xì)介紹希望你能有所收獲。
10年積累的網(wǎng)站設(shè)計(jì)、成都網(wǎng)站設(shè)計(jì)經(jīng)驗(yàn),可以快速應(yīng)對客戶對網(wǎng)站的新想法和需求。提供各種問題對應(yīng)的解決方案。讓選擇我們的客戶得到更好、更有力的網(wǎng)絡(luò)服務(wù)。我雖然不認(rèn)識你,你也不認(rèn)識我。但先網(wǎng)站制作后付款的網(wǎng)站建設(shè)流程,更有西盟免費(fèi)網(wǎng)站建設(shè)讓你可以放心的選擇與我們合作。
dubbo-2.7.3/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/dispatcher/execution/ExecutionDispatcher.java
public class ExecutionDispatcher implements Dispatcher { public static final String NAME = "execution"; @Override public ChannelHandler dispatch(ChannelHandler handler, URL url) { return new ExecutionChannelHandler(handler, url); } }
ExecutionDispatcher實(shí)現(xiàn)了Dispatcher接口,其dispatch方法返回的是ExecutionChannelHandler
dubbo-2.7.3/dubbo-remoting/dubbo-remoting-api/src/main/java/org/apache/dubbo/remoting/transport/dispatcher/execution/ExecutionChannelHandler.java
public class ExecutionChannelHandler extends WrappedChannelHandler { public ExecutionChannelHandler(ChannelHandler handler, URL url) { super(handler, url); } @Override public void received(Channel channel, Object message) throws RemotingException { ExecutorService executor = getExecutorService(); if (message instanceof Request) { try { executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message)); } catch (Throwable t) { // FIXME: when the thread pool is full, SERVER_THREADPOOL_EXHAUSTED_ERROR cannot return properly, // therefore the consumer side has to wait until gets timeout. This is a temporary solution to prevent // this scenario from happening, but a better solution should be considered later. if (t instanceof RejectedExecutionException) { Request request = (Request) message; if (request.isTwoWay()) { String msg = "Server side(" + url.getIp() + "," + url.getPort() + ") thread pool is exhausted, detail msg:" + t.getMessage(); Response response = new Response(request.getId(), request.getVersion()); response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR); response.setErrorMessage(msg); channel.send(response); return; } } throw new ExecutionException(message, channel, getClass() + " error when process received event.", t); } } else { handler.received(channel, message); } } }
ExecutionChannelHandler繼承了WrappedChannelHandler,其received方法判斷message是否是Request類型,如果是則創(chuàng)建ChannelEventRunnable放到線程池里頭執(zhí)行,如果不是則直接執(zhí)行handler.received
dubbo-2.7.3/dubbo-remoting/dubbo-remoting-api/src/test/java/org/apache/dubbo/remoting/PerformanceServerTest.java
public class PerformanceServerTest { private static final Logger logger = LoggerFactory.getLogger(PerformanceServerTest.class); private static ExchangeServer server = null; private static void restartServer(int times, int alive, int sleep) throws Exception { if (server != null && !server.isClosed()) { server.close(); Thread.sleep(100); } for (int i = 0; i < times; i++) { logger.info("restart times:" + i); server = statServer(); if (alive > 0) Thread.sleep(alive); server.close(); if (sleep > 0) Thread.sleep(sleep); } server = statServer(); } private static ExchangeServer statServer() throws Exception { final int port = PerformanceUtils.getIntProperty("port", 9911); final String transporter = PerformanceUtils.getProperty(Constants.TRANSPORTER_KEY, Constants.DEFAULT_TRANSPORTER); final String serialization = PerformanceUtils.getProperty(Constants.SERIALIZATION_KEY, Constants.DEFAULT_REMOTING_SERIALIZATION); final String threadpool = PerformanceUtils.getProperty(THREADPOOL_KEY, DEFAULT_THREADPOOL); final int threads = PerformanceUtils.getIntProperty(THREADS_KEY, DEFAULT_THREADS); final int iothreads = PerformanceUtils.getIntProperty(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS); final int buffer = PerformanceUtils.getIntProperty(BUFFER_KEY, DEFAULT_BUFFER_SIZE); final String channelHandler = PerformanceUtils.getProperty(Constants.DISPATCHER_KEY, ExecutionDispatcher.NAME); // Start server ExchangeServer server = Exchangers.bind("exchange://0.0.0.0:" + port + "?transporter=" + transporter + "&serialization=" + serialization + "&threadpool=" + threadpool + "&threads=" + threads + "&iothreads=" + iothreads + "&buffer=" + buffer + "&channel.handler=" + channelHandler, new ExchangeHandlerAdapter() { public String telnet(Channel channel, String message) throws RemotingException { return "echo: " + message + "\r\ntelnet> "; } public CompletableFuture
PerformanceServerTest的statServer方法使用PerformanceUtils.getProperty(Constants.DISPATCHER_KEY, ExecutionDispatcher.NAME)獲取channelHandler,找不到則使用ExecutionDispatcher.NAME
上述內(nèi)容就是dubbo中ExecutionDispatcher的作用是什么,你們學(xué)到知識或技能了嗎?如果還想學(xué)到更多技能或者豐富自己的知識儲備,歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。