分布式服务Dubbo+Zookeeper安全认证:KeeperErrorCode = NoAuth

一、问题背景：

我们Zookeeper正常的情况下是部署在内网进行服务注册和服务发现，难免有一些特许的情况需要部署到外网。而部署到外网的中我们需要对于服务注册和服务发现进行安全认证。至于怎样进行安全配置就不详细的描述，可以参考这篇文章（https://www.cnblogs.com/smallSevens/p/8064995.html）
在Dubbo registry上配置相应的用户、密码，服务就注册不到Zookeeper上了，会报KeeperErrorCode = NoAuth错误。

二、问题分析

代码分析

Dubbo创建注册zookeeper是通过工厂模式，而这个工厂类就是RegistryFactory，在这个工厂类中Dubbo提供了一个抽象实现类
AbstractRegistryFactory，AbstractRegistryFactory实现了getRegistry方法，并将具体的创建ZookeeperRegistry细节由子类工厂进行实现，在Dubbo使用ZookeeperRegistryFactory子类进行创建zookeeper连接和注册。

AbstractRegistryFactory.java

public Registry getRegistry(URL url) {
        url = url.setPath(RegistryService.class.getName())
                .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
                .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
        String key = url.toServiceString();
        // 锁定注册中心获取过程，保证注册中心单一实例
        LOCK.lock();
        try {
            Registry registry = REGISTRIES.get(key);
            if (registry != null) {
                return registry;
            }
            registry = createRegistry(url);
            if (registry == null) {
                throw new IllegalStateException("Can not create registry " + url);
            }
            REGISTRIES.put(key, registry);
            return registry;
        } finally {
            // 释放锁
            LOCK.unlock();
        }
    }

ZookeeperRegistryFactory.java

public class ZookeeperRegistryFactory extends AbstractRegistryFactory {
    
    private ZookeeperTransporter zookeeperTransporter;

    public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) {
        this.zookeeperTransporter = zookeeperTransporter;
    }

    public Registry createRegistry(URL url) {
        return new ZookeeperRegistry(url, zookeeperTransporter);
    }

}

我们通过ZookeeperRegistry发现zookeeper的连接是通过zookeeperTransporter进行创建。

public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
        super(url);
        if (url.isAnyHost()) {
            throw new IllegalStateException("registry address == null");
        }
        String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
        if (! group.startsWith(Constants.PATH_SEPARATOR)) {
            group = Constants.PATH_SEPARATOR + group;
        }
        this.root = group;
        zkClient = zookeeperTransporter.connect(url);
        zkClient.addStateListener(new StateListener() {
            public void stateChanged(int state) {
                if (state == RECONNECTED) {
                    try {
                        recover();
                    } catch (Exception e) {
                        logger.error(e.getMessage(), e);
                    }
                }
            }
        });
    }

zookeeperTransporter是一个接口，而该接口提供了两种实现：CuratorZookeeperTransporter和ZkclientZookeeperTransporter，而这两个实现类分别创建CuratorZookeeperClient和ZkclientZookeeperClient。

public class CuratorZookeeperTransporter implements ZookeeperTransporter {

    public ZookeeperClient connect(URL url) {
        return new CuratorZookeeperClient(url);
    }

}

public class ZkclientZookeeperTransporter implements ZookeeperTransporter {

    public ZookeeperClient connect(URL url) {
        return new ZkclientZookeeperClient(url);
    }
}

通过对CuratorZookeeperClient和ZkclientZookeeperClient的构造方法发现，ZkclientZookeeperClient是没有进行设置zookeeper的auth的账号和密码。

public ZkclientZookeeperClient(URL url) {
        super(url);
        client = new ZkClient(
                url.getBackupAddress(),
                url.getParameter(Constants.SESSION_TIMEOUT_KEY, Constants.DEFAULT_SESSION_TIMEOUT),
                url.getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_REGISTRY_CONNECT_TIMEOUT));
        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 CuratorZookeeperClient(URL url) {
        super(url);
        Builder builder = CuratorFrameworkFactory.builder()
                .connectString(url.getBackupAddress())
                .retryPolicy(new RetryNTimes(Integer.MAX_VALUE, 1000))
                .connectionTimeoutMs(url.getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_REGISTRY_CONNECT_TIMEOUT))
                .sessionTimeoutMs(url.getParameter(Constants.SESSION_TIMEOUT_KEY, Constants.DEFAULT_SESSION_TIMEOUT));
   //设置zookeeper的auth账号和密码。
        String authority = url.getAuthority();
        if (authority != null && authority.length() > 0) {
            builder = builder.authorization("digest", authority.getBytes());
        }
        client = builder.build();
        client.getConnectionStateListenable().addListener(
                new ConnectionStateListener() {
                    public void stateChanged(CuratorFramework client,
                            ConnectionState state) {
                        if (state == ConnectionState.LOST) {
                            CuratorZookeeperClient.this
                                    .stateChanged(StateListener.DISCONNECTED);
                        } else if (state == ConnectionState.CONNECTED) {
                            CuratorZookeeperClient.this
                                    .stateChanged(StateListener.CONNECTED);
                        } else if (state == ConnectionState.RECONNECTED) {
                            CuratorZookeeperClient.this
                                    .stateChanged(StateListener.RECONNECTED);
                        }
                    }
                });
        c

总结：通过源码的分析发现使用ZkclientZookeeperClient进行连接zookeeper和注册服务是不会设置安全配置。

三、解决方案

将dubbo.registry.client由zkclient改为curator即可。dubbo.registry.client = curator