dubbo源码分析-2（注册表AbstractRegistry设计技巧讲解）

 

在一些客户端多线程高并发的应用场景中，编程中会采用本地文件缓存的方式来保存一些Java类中的信息，如某个Map中的<key,value>缓存为property文件。

如果多个线程同时访问这个property文件，会造成文件版本不一致、读写内容过时等错误情况。

 

笔者在学习dubbo框架的过程中，发现了dubbo框架中有对于本地URL信息的property缓存机制（为了使得服务消费方能很好的接入服务提供方，减轻注册中心的压力）

 

1：缓存文件保存机制线程来完成

 

   dubbo中申明为内部类，这样对类使用者来说，是透明的缓存机制。

    private class SaveProperties implements Runnable{
        private long version;
        private SaveProperties(long version){
            this.version = version;
        }
        public void run() {
            doSaveProperties(version);
        }
    }

 2：文件保存过程（追加/覆盖属性配置，作为算法流程域，被动调用）

    public void doSaveProperties(long version) {
        if(version < lastCacheChanged.get()){
            return;
        }
        if (file == null) {
            return;
        }
        Properties newProperties = new Properties();
        // 保存之前先读取一遍，防止多个注册中心之间冲突
        InputStream in = null;
        try {
            if (file.exists()) {
                in = new FileInputStream(file);
                newProperties.load(in);
            }
        } catch (Throwable e) {
            logger.warn("Failed to load registry store file, cause: " + e.getMessage(), e);
        } finally {
            if (in != null) {
                try {
                    in.close();
                } catch (IOException e) {
                    logger.warn(e.getMessage(), e);
                }
            }
        }     
     // 保存
        try {
            //批量属性配置转储，类似于HashMap中的addAll机制
			newProperties.putAll(properties);
            //注意：辅助文件锁机制，创建该文件是为了方便对主文件加锁
            File lockfile = new File(file.getAbsolutePath() + ".lock");
            if (!lockfile.exists()) {
            	lockfile.createNewFile();
            }
            //使用  RandomAccessFile类即可，只需要获得文件管道句柄
            RandomAccessFile raf = new RandomAccessFile(lockfile, "rw");
            try {
                //1：获取文件句柄
                FileChannel channel = raf.getChannel();
                try {
                    //2：句柄加锁
                    FileLock lock = channel.tryLock();
                	if (lock == null) {
                        throw new IOException("Can not lock the registry cache file " + file.getAbsolutePath() + ", ignore and retry later, maybe multi java process use the file, please config: dubbo.registry.file=xxx.properties");
                    }
                    //3：进入原子操作区，操作为保存文件
                   //---------------------------------------------------
                    try {
                    	if (! file.exists()) {
                            file.createNewFile();
                        }
                        FileOutputStream outputFile = new FileOutputStream(file);  
                        try {
                            newProperties.store(outputFile, "Dubbo Registry Cache");
                        } finally {
                        	outputFile.close();
                        }
                        //4：退出原子操作区
                        //-----------------------------------------------
                    } finally {
                        //5：释放辅助文件锁
                    	lock.release();
                    }
                } finally {
                    //释放文件句柄
                    channel.close();
                }
            } finally {
                raf.close();
            }
        } catch (Throwable e) {
            if (version < lastCacheChanged.get()) {
                return;
            } else {
                registryCacheExecutor.execute(new SaveProperties(lastCacheChanged.incrementAndGet()));
            }
            logger.warn("Failed to save registry store file, cause: " + e.getMessage(), e);
        }
    }

 

3：属性文件读取（主动同步调用方法）

    private void loadProperties() {
        if (file != null && file.exists()) {
            InputStream in = null;
            try {
                in = new FileInputStream(file);
                properties.load(in);
                if (logger.isInfoEnabled()) {
                    logger.info("Load registry store file " + file + ", data: " + properties);
                }
            } catch (Throwable e) {
                logger.warn("Failed to load registry store file " + file, e);
            } finally {
                if (in != null) {
                    try {
                        in.close();
                    } catch (IOException e) {
                        logger.warn(e.getMessage(), e);
                    }
                }
            }
        }
    }

 4:属性文件保存（主动调用方法，内嵌缓存机制）

private void saveProperties(URL url) {
        if (file == null) {
            return;
        }
        try {
            StringBuilder buf = new StringBuilder();
            Map<String, List<URL>> categoryNotified = notified.get(url);
            if (categoryNotified != null) {
                for (List<URL> us : categoryNotified.values()) {
                    for (URL u : us) {
                        if (buf.length() > 0) {
                            buf.append(URL_SEPARATOR);
                        }
                        buf.append(u.toFullString());
                    }
                }
            }
            properties.setProperty(url.getServiceKey(), buf.toString());
            long version = lastCacheChanged.incrementAndGet();
            if (syncSaveFile) {
                doSaveProperties(version);
            } else {
                registryCacheExecutor.execute(new SaveProperties(version));
            }
        } catch (Throwable t) {
            logger.warn(t.getMessage(), t);
        }
    }

 

5.文件保存任务异步执行

 

 变量支持

 // 文件缓存定时写入
    private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));

    //是否是同步保存文件
    private final boolean syncSaveFile;
//使用原子变量来保证文件版本的一致性
    private final AtomicLong lastCacheChanged = new AtomicLong();

 算法流程

            //使用原子操作来保证文件版本的增量可靠性
            long version = lastCacheChanged.incrementAndGet();
            if (syncSaveFile) {
                doSaveProperties(version);
            } else {
            //启动异步定时任务执行器，将Runable丢入异步线程池中
                registryCacheExecutor.execute(new SaveProperties(version));
            }

 

 

总结：当A线程进入原子操作区间时，先对辅助文件加锁，然后操纵主文件，操作结束后，释放辅助文件锁。

如此一来，当A线程没有退出原子区域时候，B线程是无法进入进入原子区域的，因为获取不到文件锁。

这样就可以保证对主文件的操作安全性。

 

也许你会问：“为什么不直接对主文件加锁呢？”  答案是，对主文件加锁了，又如何操作主文件呢，因为文件锁的机制是会对文件的操作屏蔽的。

 

小小的一个抽象类，尽然引入了这么多的Java高端技术，不得不佩服编码者功力之深厚，优化后的类文件，代码又精简又实用，而且在高并发的应用场景，相信一定很牛的。

难怪Dubbo框架这么成熟，在淘宝内部应用也如此广泛~

 

评论

3 楼 xiangkui 2014-02-13  

引用

文件锁定以整个 Java 虚拟机来保持。但它们不适用于控制同一虚拟机内多个线程对文件的访问。

多个并发线程可安全地使用文件锁定对象。

也就是说，
１：JVM内部线程之间可以获取重复的共享锁定，操作系统进程之间不允许获取重复的独占锁(在一些特殊的操作系统中，jvm内部共享锁会转换为独占锁)

1.1 可通过调用某个锁定的 isShared 方法来确定它是独占的还是共享的

２：不适合在JVM内对文件进行访问，因为在JVM内是共享锁，尽管如此，FileLock是并发安全的

3:　判断是否锁定重叠，可使用 overlaps 方法。

2 楼 xiangkui 2014-02-13  

hengyunabc 写道

貌似LZ有个地方理解不对，FileLock是进程级的锁，而且
// 文件缓存定时写入
    private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));

这个executor是只有一个线程的。
FileLock的作用是防止其它线程来修改这个cache文件，而不是同进程内的线程。

刚查了一下，确实是进程级别的加锁，参考地址为
http://dlc.sun.com.edgesuite.net/jdk/jdk-api-localizations/jdk-api-zh-cn/publish/1.6.0/html/zh_CN/api/java/nio/channels/FileLock.html
感谢指正,

1 楼 hengyunabc 2014-02-13  

貌似LZ有个地方理解不对，FileLock是进程级的锁，而且
// 文件缓存定时写入
    private final ExecutorService registryCacheExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("DubboSaveRegistryCache", true));

这个executor是只有一个线程的。
FileLock的作用是防止其它线程来修改这个cache文件，而不是同进程内的线程。