【Dubbo】dubbo连接zookeeper注册中心因为断网导致线程无限等待问题

ゝ一纸荒年。 2022-06-18 09:15 285阅读 0赞

最近维护的系统切换了网络环境，由联通换成了电信网络，因为某些过滤规则导致系统连不上zookeeper服务器（应用系统机器在深圳，网络为电信线路，zookeeper服务器在北京，网络为联通线路），因为我不是运维人员也不懂运维相关的技术，所以排查了很久也不知道原因，最后无奈之下把深圳这边的网络切回了联通，系统恢复正常。

但是因为本次事故体现了一个很严重的问题，即当zookeeper注册中心连不上时dubbo的线程会无限等待，因为系统有一些定时任务会比较频繁地开启新线程连接dubbo，所以导致的结果是tomcat一会儿线程池就满了，其它的不依赖dubbo的功能也被阻塞无法使用。

所以需要解决一个问题，即在断网的情况下要保证应用程序可以运行（有很多功能不依赖外网），一开始我以为dubbo应该有对ZKClient连接相关的超时时间配置，结果找了很久也没发现，后来debug了dubbo的源代码发现根本就没有设置超时时间，ZKClient默认的超时时间是Integer.MAX\_VALUE，几乎等于无限等待，所以无奈之下只好重写了dubbo的ZookeeperClient实现，好在dubbo的扩展性非常好，基于SPI的扩展非常方便，下面是我的扩展代码：

1、增加一个文件com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter放置于项目的META-INF/dubbo/internal文件夹下，dubbo便会自动扫描到这个文件。文件内容很简单就一句话：zkclient=com.gwall.zookeeper.ZkclientZookeeperTransporter   
即把dubbo原来的默认实现替换为我的实现。   
2、ZkclientZookeeperTransporter并不是我想替换的代码，我要替换的是ZkclientZookeeperTransporter的成员变量ZookeeperClient，只是dubbo只在ZookeeperTransporter上加了SPI注解，所以只好这样办了，ZkclientZookeeperTransporter代码照抄过来。   
3、在ZkclientZookeeperTransporter里面使用自己的ZkclientZookeeperClient，代码如下：

package com.gwall.zookeeper;
    
    import java.util.List;
    
    import org.I0Itec.zkclient.IZkChildListener;
    import org.I0Itec.zkclient.IZkStateListener;
    import org.I0Itec.zkclient.ZkClient;
    import org.I0Itec.zkclient.exception.ZkNoNodeException;
    import org.I0Itec.zkclient.exception.ZkNodeExistsException;
    import org.apache.zookeeper.Watcher.Event.KeeperState;
    
    import com.alibaba.dubbo.common.URL;
    import com.alibaba.dubbo.remoting.zookeeper.ChildListener;
    import com.alibaba.dubbo.remoting.zookeeper.StateListener;
    import com.alibaba.dubbo.remoting.zookeeper.support.AbstractZookeeperClient;
    
    /** * 修改dubbo提供的ZkclientZookeeperClient * 主要目的是增加一个连接zookeeper的超时时间,避免ZkClient默认的无限等待 * @author long.zr * */
    public class ZkclientZookeeperClient extends AbstractZookeeperClient<IZkChildListener> {
    
        private final ZkClient client;
    
        private volatile KeeperState state = KeeperState.SyncConnected;
    
        public ZkclientZookeeperClient(URL url) {
            super(url);
            //设置超时时间为5000毫秒
            client = new ZkClient(url.getBackupAddress(),5000);
            client.subscribeStateChanges(new IZkStateListener() {
                public void handleStateChanged(KeeperState state) throws Exception {
                    ZkclientZookeeperClient.this.state = state;
                    if (state == KeeperState.Disconnected) {
                        stateChanged(StateListener.DISCONNECTED);
                    } else if (state == KeeperState.SyncConnected) {
                        stateChanged(StateListener.CONNECTED);
                    }
                }
                public void handleNewSession() throws Exception {
                    stateChanged(StateListener.RECONNECTED);
                }
            });
        }
    
        public void createPersistent(String path) {
            try {
                client.createPersistent(path, true);
            } catch (ZkNodeExistsException e) {
            }
        }
    
        public void createEphemeral(String path) {
            try {
                client.createEphemeral(path);
            } catch (ZkNodeExistsException e) {
            }
        }
    
        public void delete(String path) {
            try {
                client.delete(path);
            } catch (ZkNoNodeException e) {
            }
        }
    
        public List<String> getChildren(String path) {
            try {
                return client.getChildren(path);
            } catch (ZkNoNodeException e) {
                return null;
            }
        }
    
        public boolean isConnected() {
            return state == KeeperState.SyncConnected;
        }
    
        public void doClose() {
            client.close();
        }
    
        public IZkChildListener createTargetChildListener(String path, final ChildListener listener) {
            return new IZkChildListener() {
                public void handleChildChange(String parentPath, List<String> currentChilds)
                        throws Exception {
                    listener.childChanged(parentPath, currentChilds);
                }
            };
        }
    
        public List<String> addTargetChildListener(String path, final IZkChildListener listener) {
            return client.subscribeChildChanges(path, listener);
        }
    
        public void removeTargetChildListener(String path, IZkChildListener listener) {
            client.unsubscribeChildChanges(path,  listener);
        }
    
    }

框架