`

Dubbo不能优雅停机,导致停止服务的时候,业务掉单

阅读更多

Dubbo 优雅停机修改方案

 

1.      服务端不能优雅停机的原因:

NettyServer在构造函数中会调用

ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME))

方法将handler进行包装,包装成MultiMessageHandler的一个对象。在下面红色代码中会判断handler是否是WrappedChannelHandler对象,只有是的时候才会对executor对象复值。因为MultiMessageHandler对象不是WrappedChannelHandler的子类,所以executor为空。

NettyServerclose方法被调用的时候,会调用ExecutorUtil.gracefulShutdown方法,gracefulShutdown方法只有executor不为空时才会等待线程池关闭

 

public abstract class AbstractServer extends AbstractEndpoint implements Server {

   public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {

        super(url, handler);

        localAddress = getUrl().toInetSocketAddress();

        String host = url.getParameter(Constants.ANYHOST_KEY, false)

                        || NetUtils.isInvalidLocalHost(getUrl().getHost())

                        ? NetUtils.ANYHOST : getUrl().getHost();

        bindAddress = new InetSocketAddress(host, getUrl().getPort());

        this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);

        this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);

        try {

            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);

        }

        if (handler instanceof WrappedChannelHandler ){

            executor = ((WrappedChannelHandler)handler).getExecutor();

        }

    }

  public void close(int timeout) {

        ExecutorUtil.gracefulShutdown(executor ,timeout);

        close();

    }

}

 

 

2. 服务端改动:

修改了AbstractServer类的红色部分

 public abstract class AbstractServer extends AbstractEndpoint implements Server {

 

    private static final Logger logger = LoggerFactory.getLogger(AbstractServer.class);

 

    private InetSocketAddress              localAddress;

 

    private InetSocketAddress              bindAddress;

 

    private int                            accepts;

 

    private int                            idleTimeout = 600; //600 seconds

 

    protected static final String SERVER_THREAD_POOL_NAME  ="DubboServerHandler";

 

    ExecutorService executor;

 

    public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {

        super(url, handler);

        localAddress = getUrl().toInetSocketAddress();

        String host = url.getParameter(Constants.ANYHOST_KEY, false)

                || NetUtils.isInvalidLocalHost(getUrl().getHost())

                ? NetUtils.ANYHOST : getUrl().getHost();

        bindAddress = new InetSocketAddress(host, getUrl().getPort());

        this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);

        this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);

        try {

            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);

        }

        try {

            recycleToGetExecutor(handler);

        } catch (Exception e) {

            logger.error("recycle to get executor failed");

        }

    }

 

    private void recycleToGetExecutor(ChannelHandler handler) throws NoSuchFieldException, IllegalAccessException {

        if(handler == null)

        {

            return;

        }

 

        if (handler instanceof WrappedChannelHandler ){

            executor = ((WrappedChannelHandler)handler).getExecutor();

        }else if(handler instanceof AbstractChannelHandlerDelegate){

            Field field = AbstractChannelHandlerDelegate.class.getDeclaredField("handler");

            field.setAccessible(true);

            recycleToGetExecutor((ChannelHandler) field.get(handler));

        }

    }

 

    protected abstract void doOpen() throws Throwable;

 

    protected abstract void doClose() throws Throwable;

 

    public void reset(URL url) {

        if (url == null) {

            return;

        }

        try {

            if (url.hasParameter(Constants.ACCEPTS_KEY)) {

                int a = url.getParameter(Constants.ACCEPTS_KEY, 0);

                if (a > 0) {

                    this.accepts = a;

                }

            }

        } catch (Throwable t) {

            logger.error(t.getMessage(), t);

        }

        try {

            if (url.hasParameter(Constants.IDLE_TIMEOUT_KEY)) {

                int t = url.getParameter(Constants.IDLE_TIMEOUT_KEY, 0);

                if (t > 0) {

                    this.idleTimeout = t;

                }

            }

        } catch (Throwable t) {

            logger.error(t.getMessage(), t);

        }

        try {

            if (url.hasParameter(Constants.THREADS_KEY)

                    && executor instanceof ThreadPoolExecutor && !executor.isShutdown()) {

                ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;

                int threads = url.getParameter(Constants.THREADS_KEY, 0);

                int max = threadPoolExecutor.getMaximumPoolSize();

                int core = threadPoolExecutor.getCorePoolSize();

                if (threads > 0 && (threads != max || threads != core)) {

                    if (threads < core) {

                        threadPoolExecutor.setCorePoolSize(threads);

                        if (core == max) {

                            threadPoolExecutor.setMaximumPoolSize(threads);

                        }

                    } else {

                        threadPoolExecutor.setMaximumPoolSize(threads);

                        if (core == max) {

                            threadPoolExecutor.setCorePoolSize(threads);

                        }

                    }

                }

            }

        } catch (Throwable t) {

            logger.error(t.getMessage(), t);

        }

        super.setUrl(getUrl().addParameters(url.getParameters()));

    }

 

    public void send(Object message, boolean sent) throws RemotingException {

        Collection<Channel> channels = getChannels();

        for (Channel channel : channels) {

            if (channel.isConnected()) {

                channel.send(message, sent);

            }

        }

    }

 

    public void close() {

        if (logger.isInfoEnabled()) {

            logger.info("Close " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());

        }

        ExecutorUtil.shutdownNow(executor ,100);

        try {

            super.close();

        } catch (Throwable e) {

            logger.warn(e.getMessage(), e);

        }

        try {

            doClose();

        } catch (Throwable e) {

            logger.warn(e.getMessage(), e);

        }

    }

 

    public void close(int timeout) {

        ExecutorUtil.gracefulShutdown(executor ,timeout);

        close();

    }

 

    public InetSocketAddress getLocalAddress() {

        return localAddress;

    }

 

    public InetSocketAddress getBindAddress() {

        return bindAddress;

    }

 

    public int getAccepts() {

        return accepts;

    }

 

    public int getIdleTimeout() {

        return idleTimeout;

    }

 

    @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);

    }

 

    @Override

    public void disconnected(Channel ch) throws RemotingException {

        Collection<Channel> channels = getChannels();

        if (channels.size() == 0){

            logger.warn("All clients has discontected from " + ch.getLocalAddress() + ". You can graceful shutdown now.");

        }

        super.disconnected(ch);

    }

 

}

 

3.   客户端不能优雅停机的原因:

DubboInvoker停机的时候直接调用了client.close()方法应该调用client.close(long timeout)方法

public class DubboInvoker<T> extends AbstractInvoker<T> {

public void destroy() {

        //防止client被关闭多次.connect per jvm的情况下,client.close方法会调用计数器-1,当计数器小于等于0的情况下,才真正关闭

        if (super.isDestroyed()){

            return ;

        } else {

            //dubbo check ,避免多次关闭

            destroyLock.lock();

            try{

                if (super.isDestroyed()){

                    return ;

                }

                super.destroy();

                if (invokers != null){

                    invokers.remove(this);

                }

                for (ExchangeClient client : clients) {

                    try {

                        client.close();

                    } catch (Throwable t) {

                        logger.warn(t.getMessage(), t);

                    }

                }

               

            }finally {

                destroyLock.unlock();

            }

        }

}

}

 

HeaderExchangeChannel类中判断HeaderExchangeChannel.this是否在DefaultFuture中。实际上在处理请求的时候是将HeaderExchangeChannel对象中channel成员变量放到DefaultFuture中的

final class HeaderExchangeChannel implements ExchangeChannel {

  public ResponseFuture request(Object request, int timeout) throws RemotingException {

        if (closed) {

            throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");

        }

        // create request.

        Request req = new Request();

        req.setVersion("2.0.0");

        req.setTwoWay(true);

        req.setData(request);

        DefaultFuture future = new DefaultFuture(channel, req, timeout);

        try{

            channel.send(req);

        }catch (RemotingException e) {

            future.cancel();

            throw e;

        }

        return future;

}

 

public void close(int timeout) {

        if (closed) {

            return;

        }

        closed = true;

        if (timeout > 0) {

            long start = System.currentTimeMillis();

            while (DefaultFuture.hasFuture(HeaderExchangeChannel.this)

                    && System.currentTimeMillis() - start < timeout) {

                try {

                    Thread.sleep(10);

                } catch (InterruptedException e) {

                    logger.warn(e.getMessage(), e);

                }

            }

        }

        close();

    }

}

 

4.   客户端的修改

 

修改下面高量代码:

public class DubboInvoker<T> extends AbstractInvoker<T> {

    public void destroy() {

        //防止client被关闭多次.connect per jvm的情况下,client.close方法会调用计数器-1,当计数器小于等于0的情况下,才真正关闭

        if (super.isDestroyed()){

            return ;

        } else {

            //dubbo check ,避免多次关闭

            destroyLock.lock();

            try{

                if (super.isDestroyed()){

                    return ;

                }

                super.destroy();

                if (invokers != null){

                    invokers.remove(this);

                }

                for (ExchangeClient client : clients) {

                    try {

                        logger.info("start to shutdown:" + getServerShutdownTimeout());

                        client.close(getServerShutdownTimeout());

                    } catch (Throwable t) {

                        logger.warn(t.getMessage(), t);

                    }

                }

 

            }finally {

                destroyLock.unlock();

            }

        }

    }

 

    protected static int getServerShutdownTimeout() {

        int timeout = Constants.DEFAULT_SERVER_SHUTDOWN_TIMEOUT;

        String value = ConfigUtils.getProperty(Constants.SHUTDOWN_WAIT_KEY);

        if (value != null && value.length() > 0) {

            try{

                timeout = Integer.parseInt(value);

            }catch (Exception e) {

            }

        } else {

            value = ConfigUtils.getProperty(Constants.SHUTDOWN_WAIT_SECONDS_KEY);

            if (value != null && value.length() > 0) {

                try{

                    timeout = Integer.parseInt(value) * 1000;

                }catch (Exception e) {

                }

            }

        }

 

        return timeout;

    }

}

 

final class HeaderExchangeChannel implements ExchangeChannel {

     // graceful close

    public void close(int timeout) {

        if (closed) {

            return;

        }

        closed = true;

        if (timeout > 0) {

            long start = System.currentTimeMillis();

            while (DefaultFuture.hasFuture(channel)

                    && System.currentTimeMillis() - start < timeout) {

                try {

                    Thread.sleep(10);

                } catch (InterruptedException e) {

                    logger.warn(e.getMessage(), e);

                }

            }

        }

        close();

    }

}

 

 

注意服务端和客户端都需要加上配置dubbo.service.shutdown.wait=60000,设定停机等待时间

分享到:
评论
2 楼 无红墙 2017-08-08  
另一种修改,请参考:
https://github.com/taige/dubbo/commit/ed08ffac5cfacd0d226299ce7f7633057620bec4
1 楼 fish_no7 2016-03-16  

if (handler instanceof WrappedChannelHandler ){
            executor = ((WrappedChannelHandler)handler).getExecutor();
        }

你好,请问NettyServer的话默认
public NettyServer(URL url, ChannelHandler handler) throws RemotingException{
        super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
    }

wrap的都是MultiMessageHandler,包括minaServer

他在优雅停机的时候为什么又要是WrappedChannelHandler 的子类才给优雅停机呢,是BUG么

相关推荐

Global site tag (gtag.js) - Google Analytics