首页 > 资讯 > 后端开发 > Python >JavaDubbo协议下的服务端线程使用详解

507

分享到

JavaDubbo协议下的服务端线程使用详解

Java Dubbo服务端线程 Java服务端线程模型 2023-03-01 11:03:36 507人浏览八月长安

Python 官方文档：入门教程 => 点击学习

摘要

目录Provider端线程模型AllDispatcherDirectDispatcherExecutionDispatcherMessageOnlyDispatcherConnect

Provider端线程模型

在了解服务端线程模型之前，先了解一下dubbo对Channel上的操作抽象，Dubbo将Channel上的操作成了5中行为，分别是：建立连接、断开连接、发送消息、接收消息、异常捕获，Channel上的操作的接口为org.apache.dubbo.remoting.ChannelHandler，该接口是SPI的，用户可以自己扩展，接口代码如下：

该接口抽象的五种Channel上的行为解释如下：

建立连接：connected，主要是的职责是在channel记录read、write的时间，以及处理建立连接后的回调逻辑，比如dubbo支持在断开后自定义回调的hook（onconnect），即在该操作中执行。
断开连接：disconnected，主要是的职责是在channel移除read、write的时间，以及处理端开连接后的回调逻辑，比如dubbo支持在断开后自定义回调的hook（ondisconnect），即在该操作中执行。
发送消息：sent，包括发送请求和发送响应。记录write的时间。
接收消息：received，包括接收请求和接收响应。记录read的时间。
异常捕获：caught，用于处理在channel上发生的各类异常。

Dubbo框架的线程模型与以上这五种行为息息相关，Dubbo协议Provider端线程模型提供了五种实现，虽说都是五种但是别把二者混淆，线程模型的顶级接口是org.apache.dubbo.remoting.Dispatcher，该接口也是SPI的，提供的五种实现分别是AllDispatcher、DirectDispatcher、MessageOnlyDispatcher、ExecutionDispatcher、ConnectionOrderedDispatcher，默认的使用的是AllDispatcher。

org.apache.dubbo.remoting.ChannelHandler作为Channel上的行为的顶级接口对应Dubbo协议Provider端的5种线程模型同样也提供了对应的5种实现，分别是AllChannelHandler、DirectChannelHandler、MessageOnlyChannelHandler、ExecutionChannelHandler、ConnectionOrderedChannelHandler，这里Channel上行为的具体实现不展开讨论。

Channel上行为和线程模型之间使用策略可以参考org.apache.dubbo.remoting.transport.dispatcher.ChannelHandlers的源代码，这里不做详细的介绍，下面的各个章节只针对5种线程模型做简单的介绍。

AllDispatcher

IO线程上的操作：

接口响应序列化
sent操作

Dubbo线程池上的操作：

received、connected、disconnected、caught都是在Dubbo线程池上执行
服务端反序列化操作的Dubbo线程池上执行

AllDispatcher代码如下，AllDispatcher的dispatch方法实例化了AllChannelHandler，AllChannelHandler实现了received、connected、disconnected、caught操作在dubbo线程池中，代码如下：

public class AllDispatcher implements Dispatcher {
    public static final String NAME = "all";
    @Override
    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return new AllChannelHandler(handler, url);
    }
}

public class AllChannelHandler extends WrappedChannelHandler {
    public AllChannelHandler(ChannelHandler handler, URL url) {
        super(handler, url);
    }
    @Override
    public void connected(Channel channel) throws RemotingException {
        ExecutorService executor = getSharedExecutorService();
        try {
            executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
        } catch (Throwable t) {
            throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t);
        }
    }
    @Override
    public void disconnected(Channel channel) throws RemotingException {
        ExecutorService executor = getSharedExecutorService();
        try {
            executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED));
        } catch (Throwable t) {
            throw new ExecutionException("disconnect event", channel, getClass() + " error when process disconnected event .", t);
        }
    }
    @Override
    public void received(Channel channel, Object message) throws RemotingException {
        ExecutorService executor = getPreferredExecutorService(message);
        try {
            executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
        } catch (Throwable t) {
            if(message instanceof Request && t instanceof RejectedExecutionException){
                sendFeedback(channel, (Request) message, t);
                return;
            }
            throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
        }
    }
    @Override
    public void caught(Channel channel, Throwable exception) throws RemotingException {
        ExecutorService executor = getSharedExecutorService();
        try {
            executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CAUGHT, exception));
        } catch (Throwable t) {
            throw new ExecutionException("caught event", channel, getClass() + " error when process caught event .", t);
        }
    }
}

DirectDispatcher

该线程模型Channel上的所有行为均在IO线程中执行，并没有在Dubbo线程池中执行

DirectDispatcher与AllDispatcher相似，实例化了DirectChannelHandler，DirectChannelHandler只实现了received行为，但是received中获取的线程池如果是ThreadlessExecutor才会提交task，否则也是在ChannelHandler中执行received行为，ThreadlessExecutor和普通线程池最大的区别是不会管理任何线程，这里不展开讨论。

public class DirectDispatcher implements Dispatcher {
    public static final String NAME = "direct";
    @Override
    public ChannelHandler dispatch(ChannelHandler handler, URL url) {
        return new DirectChannelHandler(handler, url);
    }
}

public class DirectChannelHandler extends WrappedChannelHandler {
    public DirectChannelHandler(ChannelHandler handler, URL url) {
        super(handler, url);
    }
    @Override
    public void received(Channel channel, Object message) throws RemotingException {
        ExecutorService executor = getPreferredExecutorService(message);
        if (executor instanceof ThreadlessExecutor) {
            try {
                executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
            } catch (Throwable t) {
                throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
            }
        } else {
            handler.received(channel, message);
        }
    }
}

ExecutionDispatcher

在IO线程中执行的操作有：

sent、connected、disconnected、caught操作在IO线程上执行。
序列化响应在IO线程上执行。

在Dubbo线程中执行的操作有：

received都是在Dubbo线程上执行的。
反序列化请求的行为在Dubbo中做的。

同样的，我们可以直接看ExecutionChannelHandler源码，逻辑是当message的类型是Request时received行为在Dubbo线程池执行。感兴趣的可以自己看源码，这里不做介绍。

MessageOnlyDispatcher

Message Only Dispatcher所有的received行为和反序列化都是在dubbo线程池中执行的

public class MessageOnlyChannelHandler extends WrappedChannelHandler {
    public MessageOnlyChannelHandler(ChannelHandler handler, URL url) {
        super(handler, url);
    }
    @Override
    public void received(Channel channel, Object message) throws RemotingException {
        ExecutorService executor = getPreferredExecutorService(message);
        try {
            executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
        } catch (Throwable t) {
            if(message instanceof Request && t instanceof RejectedExecutionException){
                sendFeedback(channel, (Request) message, t);
                return;
            }
            throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
        }
    }
}

ConnectionOrderedDispatcher

该线程模型与AllDispatcher类似，sent操作和相应的序列化是在IO线程上执行；connected、disconnected、received、caught操作在dubbo线程池上执行，他们的区别是在connected、disconnected行为上ConnectionOrderedDispatcher做了线程池隔离，并且在Dubbo connected thread pool中提供了链接限制、告警灯能力，我们直接看ConnectionOrderedChannelHandler源码，代码如下：

public class ConnectionOrderedChannelHandler extends WrappedChannelHandler {
    protected final ThreadPoolExecutor connectionExecutor;
    private final int queueWarningLimit;
    public ConnectionOrderedChannelHandler(ChannelHandler handler, URL url) {
        super(handler, url);
        String threadName = url.getParameter(THREAD_NAME_KEY, DEFAULT_THREAD_NAME);
        connectionExecutor = new ThreadPoolExecutor(1, 1,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>(url.getPositiveParameter(CONNECT_QUEUE_CAPACITY, Integer.MAX_VALUE)),
                new NamedThreadFactory(threadName, true),
                new AbortPolicyWithReport(threadName, url)
        );  // FIXME There's no place to release connectionExecutor!
        queueWarningLimit = url.getParameter(CONNECT_QUEUE_WARNING_SIZE, DEFAULT_CONNECT_QUEUE_WARNING_SIZE);
    }
    @Override
    public void connected(Channel channel) throws RemotingException {
        try {
            checkQueueLength();
            connectionExecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
        } catch (Throwable t) {
            throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t);
        }
    }
    @Override
    public void disconnected(Channel channel) throws RemotingException {
        try {
            checkQueueLength();
            connectionExecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED));
        } catch (Throwable t) {
            throw new ExecutionException("disconnected event", channel, getClass() + " error when process disconnected event .", t);
        }
    }
    @Override
    public void received(Channel channel, Object message) throws RemotingException {
        ExecutorService executor = getPreferredExecutorService(message);
        try {
            executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
        } catch (Throwable t) {
            if (message instanceof Request && t instanceof RejectedExecutionException) {
                sendFeedback(channel, (Request) message, t);
                return;
            }
            throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
        }
    }
    @Override
    public void caught(Channel channel, Throwable exception) throws RemotingException {
        ExecutorService executor = getSharedExecutorService();
        try {
            executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CAUGHT, exception));
        } catch (Throwable t) {
            throw new ExecutionException("caught event", channel, getClass() + " error when process caught event .", t);
        }
    }
    private void checkQueueLength() {
        if (connectionExecutor.getQueue().size() > queueWarningLimit) {
            logger.warn(new IllegalThreadStateException("connectionordered channel handler queue size: " + connectionExecutor.getQueue().size() + " exceed the warning limit number :" + queueWarningLimit));
        }
    }
}

到此这篇关于Java Dubbo协议下的服务端线程使用详解的文章就介绍到这了,更多相关Java Dubbo服务端线程内容请搜索编程网以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程网！

您可能感兴趣的文档:

--结束END--

本文标题: JavaDubbo协议下的服务端线程使用详解

本文链接: https://lsjlt.com/news/197983.html(转载时请注明来源链接)

有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

回答

如何调试操作系统的错误？
操作系统

2023-11-15发布

回答

操作系统中的I/O系统是如何实现的？
操作系统

2023-11-15发布

回答

如何实现操作系统的内存管理？
操作系统

2023-11-15发布

回答

什么是虚拟内存，它对操作系统有什么影响？
操作系统

2023-11-15发布

回答

ASP中的MVC架构和WebForms架构有什么区别和使用场景？
ASP.NET

2023-11-15发布

回答

ASP中的数据验证和数据校验有什么不同？
ASP.NET

2023-11-15发布

回答

ASP中的ADO对象和DAO对象有什么区别和使用方法？
ASP.NET

2023-11-15发布

回答

Node.js中的包管理器NPM是什么？如何使用它进行依赖管理？
node.js

2023-11-15发布

回答

Vue.js中的动态组件是什么？如何使用它来动态渲染组件？
VUE

2023-11-15发布

回答

如何使用Vue.js实现懒加载和预加载？
VUE

2023-11-15发布

JavaDubbo协议下的服务端线程使用详解

目录

Provider端线程模型

AllDispatcher

DirectDispatcher

ExecutionDispatcher

MessageOnlyDispatcher

ConnectionOrderedDispatcher

JavaDubbo协议下的服务端线程使用详解

Java Dubbo协议下的服务端线程如何使用

解读MySQL的客户端和服务端协议

阿里云服务器托管协议下载详解

美国服务器使用TCP协议的端口有哪些

如何使用加密的远程终端协议（RDP）保护CentOS服务器

obix协议在java中的配置和使用详解

美国服务器使用TCP协议的常见端口有哪些

Golang使用http协议实现心跳检测程序过程详解

PHP使用POP3协议接收邮件的详细过程

阿里云ECS服务器465端口详解及使用建议

ip使用不受支持的协议，客户端和服务器不支持常用的SSL协议版本或密码套件。

PHP使用IMAP协议接收邮箱邮件的详细流程

kotlin之协程的理解与使用详解

服务器使用BGP协议的优势是什么

亲自教你在netty中使用TCP协议请求DNS服务器的详细过程

高防服务器使用的TCP协议4种端口分别有什么作用

PHP使用SMTP协议和PEAR库发送邮件的详细流程

Python内置的HTTP协议服务器SimpleHTTPServer使用指南

Android使用http协议与服务器通信的实例

python分析数据的方法是什么

如何使用Python实现抽奖小程序

python copy函数的作用是什么

python ffmpeg模块怎么安装和使用

python进程池创建队列的方法是什么

python无法运行文件的原因有哪些

python can't open file报错怎么解决

python keyerror错误怎么解决

python字符串处理与应用的方法有哪些

python全局变量如何定义