1. 容灾读的部分不必说。对象原本在存入缓存时就依据类型分了区域，读的时候直接在对应的区域取出就可以。

   容灾写的话，针对同一类型对象的写操作，怎样将这些对象集合在一块，待DataCore恢复后，再将容灾写过的对象覆盖到DB。

2. 缓存的写操作必须是线程安全的

缓存的底层储存机制就是基于分散的hash。

这样我们的就不须要那么大的连续空间了。示意图例如以下：

示意图例如以下：

//index对象
public class DisasterIndexDO implements Serializable {
     
private static final long serialVersionUID = -8688243351154917184L;
     public int namespace;
     public int beginIndex;
     public int currentIndex;
     public long expireLockTime;
     public static final long DEFAULT_EXPIRE_TIME = 50; // 当序列被锁时间超时，防止死锁
     
     public DisasterIndexDO(int namespace, int bIndex, int cIndex, long expireLockTime) {
        this.namespace = namespace;
        this.beginIndex = bIndex;
        this.currentIndex = cIndex;
        this.expireLockTime = expireLockTime;
     }
}

//容灾实现类
public class DisasterCacheHandler {
     RemoteCache remoteCache;

     private  final int DS_CACHE_NAMESPACE = 67; 
     private  final int DS_WRITE_REPETECOUNT = 3;
     private  final int DISASTER_INDEX = 250;
     private  final String DISASTER_KEYS = "disaster_keys";
//容灾读
    public Object dsGetRemoteData(int namespace, String key){
         return remoteCache.get(DS_CACHE_NAMESPACE, namespace+key);
    }
    
    //本地同步的namespace
     private Map<Integer, Object> synNamespace = new 
           HashMap<Integer, Object>();
     //容灾写
     protected boolean dsWriteRemoteData(int namespace, 
          String key, Serializable value) {

          //先把数据写入缓存
          remoteCache.put(this.DS_CACHE_NAMESPACE, namespace+key, value);
          //本地同步的NameSpace
          if (!this.synNamespace.containsKey(Integer.valueOf(namespace))) {
                this.synNamespace.put(Integer.valueOf(namespace), namespace);
          }
          // update the Namespace Index and namespace disaster key queue
          synchronized (this.synNamespace.get(namespace)){
               int count = this.DS_WRITE_REPETECOUNT;
               DisasterIndexDO index = null;
               do {
                    count--;
                     // try to lock Namespace Index
                     int rc = remoteCache.lock(this.DISASTER_INDEX, 
                         String.valueOf(namespace));
                    // Namespace Index not exist
                    if ( rc == 2 ) {
                         // Initialize Namespace Index
                         index = new DisasterIndexDO(namespace, 1, 1, 
                              System.currentTimeMillis());
                         remoteCache.put(this.DISASTER_INDEX, 
                              String.valueOf(namespace), index);
                 // for each namespace should handle disaster, keep the only index object
                         remoteCache.put(namespace, 
                              this.DISASTER_KEYS + index.currentIndex, key);
                         return true;
                    } else if (rc == 0) { // lock failure
                         index = (DisasterIndexDO)remoteCache.
                                   get(this.DISASTER_INDEX, String.valueOf(namespace));
                    // 假设鎖已經超時。則解开，避免在訪问缓存时死住的情况
                         if (System.currentTimeMillis() - index.expireLockTime >           
                                   DisasterIndexDO.DEFAULT_EXPIRE_TIME) {
                                        remoteCache.unLock(this.DISASTER_INDEX, 
                                             String.valueOf(namespace));
                                   continue;
                          }
                         continue;
                   } else if (rc == 1) { // lock success
                         try {
                              index = (DisasterIndexDO)remoteCache.get(this.DISASTER_INDEX,  
                                        String.valueOf(namespace));
                              // update locked Namespace Index
                              int curIdx = index.currentIndex + 1;
                              remoteCache.delete(this.DISASTER_INDEX, 
                                        String.valueOf(namespace));
                              remoteCache.put(this.DISASTER_INDEX, String.valueOf(namespace), 
                                   new DisasterIndexDO(namespace, index.beginIndex, curIdx, 
                                        System.currentTimeMillis()));
                              // keep key of this Namespace with current index
                              remoteCache.put(namespace, this.DISASTER_KEYS + curIdx, key);
                              return true;
                         }  catch (Throwable e ) {
                         } finally {
                              // unlock and handle unlock failure 
                              remoteCache.unLock(this.DISASTER_INDEX, 
                                   String.valueOf(namespace));
                         }
                   }
             } while (count >= 0);
            if (count <= 0) {
                 return false;
            }
        }
    }
}