栏目分类:
子分类:
返回
名师互学网用户登录
快速导航关闭
当前搜索
当前分类
子分类
实用工具
热门搜索
名师互学网 > IT > 前沿技术 > 大数据 > 大数据系统

线程池的实现原理(线程池默认多少个线程)

线程池的实现原理(线程池默认多少个线程)

课程主要内容:

掌握Apache Dubbo线程模型异同掌握Apache Dubbo线程池异同与选择了解Apache Dubbo线程数选择

线程模型
线程模型线程模型作用
All所有消息都派发到线程池,包括请求、响应、连接事件、断开事件、心跳等
Direct所有请求都不派发到线程池,全部在IO线程上直接执行
Message只有请求响应消息派发到线程池
Excution只有请求消息派发到线程池,不含响应
Connection在IO线程上,将连接断开事件放入队列,有序逐个执行
属性配置:

源码剖析

在NettyServer与NettyClient的构造方法中都有调用 ChannelHandlers.wrap方法

    //org.apache.dubbo.remoting.transport.netty4.NettyServer#NettyServer
    public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
        // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
        // the handler will be wrapped: MultiMessageHandler(多消息处理)->HeartbeatHandler(心跳处理)->handler
        super(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME), ChannelHandlers.wrap(handler, url));
    }

	//org.apache.dubbo.remoting.transport.netty4.NettyClient#NettyClient
    public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
    	// you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
    	// the handler will be wrapped: MultiMessageHandler->HeartbeatHandler->handler
    	super(url, wrapChannelHandler(url, ChannelHandlers.wrap(handler, url)));
    }

如下是 ChannelHandlers.wrap方法:

    //
    protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
    	//此处获取到配种中的dispatcher属性 调用对应实现的dispatcher的 dispatch方法,每一种dispatcher都对应自己的ChannelHandler
        return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
                .getAdaptiveExtension().dispatch(handler, url)));
    }

因为NettyServer与NettyClient的实现基本一直,所以我们以NettyServer为例进行讲解:

关于何时使用的 我们回到 NettyTransporter 的 bind方法中,(此处由org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#createServer方法调用过来的)

    @Override
    public RemotingServer bind(URL url, ChannelHandler handler) throws RemotingException {
        //创建NettyServer
        return new NettyServer(url, handler);
    }
    public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
        // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
        // the handler will be wrapped: MultiMessageHandler->HeartbeatHandler->handler
        //ChannelHandlers.wrap(handler, url) 方法获取到对应的 handle,并 调用父类构造方法
        super(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME), ChannelHandlers.wrap(handler, url));
    }
    public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
        //此处不再深入,此处为继续调用父类,一直到AbstractPeer类的构造方法中,将ChannelHandler handler属性赋值
        super(url, handler);
        localAddress = getUrl().toInetSocketAddress();

        String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
        int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
        if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
            bindIp = ANYHOST_VALUE;
        }
        bindAddress = new InetSocketAddress(bindIp, bindPort);
        this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
        this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT);
        try {
            //调用NettyServer的doOpen方法,进入
            doOpen();
            if (logger.isInfoEnabled()) {
                logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
            }
        } catch (Throwable t) {
            throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
                    + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
        }
        executor = executorRepository.createExecutorIfAbsent(url);
    }
    @Override
    protected void doOpen() throws Throwable {
        bootstrap = new ServerBootstrap();

        bossGroup = NettyEventLoopFactory.eventLoopGroup(1, "NettyServerBoss");
        workerGroup = NettyEventLoopFactory.eventLoopGroup(
                getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS),
                "NettyServerWorker");

        //此处会创建Netty所使用的的 Handler 即:NettyServerHandler,类中属性ChannelHandler handler持有了本类,本类实现了ChannelHandler,在父类AbstractPeer中 重写了 连接、取消连接、请求、响应 等一些请求,也是直接调用传入的对应的handle来实现的
        final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this);
        channels = nettyServerHandler.getChannels();

        bootstrap.group(bossGroup, workerGroup)
                .channel(NettyEventLoopFactory.serverSocketChannelClass())
                .option(ChannelOption.SO_REUSEADDR, Boolean.TRUE)
                .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE)
                .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                .childHandler(new ChannelInitializer() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        // FIXME: should we use getTimeout()?
                        int idleTimeout = UrlUtils.getIdleTimeout(getUrl());
                        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this);
                        if (getUrl().getParameter(SSL_ENABLED_KEY, false)) {
                            ch.pipeline().addLast("negotiation",
                                    SslHandlerInitializer.sslServerHandler(getUrl(), nettyServerHandler));
                        }
                        ch.pipeline()
                                .addLast("decoder", adapter.getDecoder())
                                .addLast("encoder", adapter.getEncoder())
                                .addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS))
                                //此处将 nettyServerHandler加入到pipeline中,用于具体业务的执行调用
                                .addLast("handler", nettyServerHandler);
                    }
                });
        // bind
        ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
        channelFuture.syncUninterruptibly();
        channel = channelFuture.channel();

    }
转载请注明:文章转载自 www.mshxw.com
本文地址:https://www.mshxw.com/it/771462.html
我们一直用心在做
关于我们 文章归档 网站地图 联系我们

版权所有 (c)2021-2022 MSHXW.COM

ICP备案号:晋ICP备2021003244-6号