Memcached 集群,客户端自动Hash到不同服务器的实现

    技术2023-03-19  28

    首先分析一下Java client 启动时的部分代码 Memcached 支持直接设置多个servers属性 来实现多个memcahced均衡,对应还有一个属性是weights,字面意思就是权重,分析了一下代码,和我想的是一样的 启动memcached的代码通常是这样的 

    Java代码  SockIOPool pool = SockIOPool.getInstance(poolname);  pool.setServers(servers);  pool.setWeights(weights);  pool.setInitConn(initConn);  pool.setMinConn(minConn);  pool.setMaxConn(maxConn);  pool.setMaxIdle(maxIdle);  pool.setMaxBusyTime(maxBusyTime);  pool.setMaintSleep(maintSleep);  pool.setSocketTO(socketTO);  pool.setSocketConnectTO(socketConnectTO);  pool.setNagle(nagle);  pool.setHashingAlg(SockIOPool.NEW_COMPAT_HASH);  pool.initialize();  MemCachedClient client = new MemCachedClient(poolname);   servers 和 weights 都是一个数组,就是说可以同时设置多个server  然后看一下 pool.initialize() 做了什么  Java代码  availPool   = new HashMap<String,Map<SockIO,Long>>( servers.length * initConn );  busyPool    = new HashMap<String,Map<SockIO,Long>>( servers.length * initConn );  deadPool    = new IdentityHashMap<SockIO,Integer>();  hostDeadDur = new HashMap<String,Long>();  hostDead    = new HashMap<String,Date>();  maxCreate   = (poolMultiplier > minConn) ? minConn : minConn / poolMultiplier;   // only create up to maxCreate connections at once  if ( log.isDebugEnabled() ) {      log.debug( "++++ initializing pool with following settings:" );      log.debug( "++++ initial size: " + initConn );      log.debug( "++++ min spare   : " + minConn );      log.debug( "++++ max spare   : " + maxConn );  }  // if servers is not set, or it empty, then  // throw a runtime exception  if ( servers == null || servers.length <= 0 ) {      log.error( "++++ trying to initialize with no servers" );      throw new IllegalStateException( "++++ trying to initialize with no servers" );  }      // initalize our internal hashing structures          if ( this.hashingAlg == CONSISTENT_HASH )          populateConsistentBuckets();      else              populateBuckets();   看到这里就是开辟一些连接池的空间,然后调用了根据我们选择的hash 算法 执行populateBuckets();或者populateConsistentBuckets();  hash算法共有4种  Java代码  // native String.hashCode();  public static final int NATIVE_HASH     = 0;      // original compatibility hashing algorithm (works with other clients)  public static final int OLD_COMPAT_HASH = 1;  // new CRC32 based compatibility hashing algorithm (works with other clients)     public static final int NEW_COMPAT_HASH = 2;  // MD5 Based -- Stops thrashing when a server added or removed  public static final int CONSISTENT_HASH = 3;           我们通常用的是 NEW_COMPAT_HASH,这个保证可以wokrs with other clients  所以看一下populateBuckets()做了什么  Java代码  this.buckets = new ArrayList<String>();  for ( int i = 0; i < servers.length; i++ ) {    if ( this.weights != null && this.weights.length > i ) {       for ( int k = 0; k < this.weights[i].intValue(); k++ ) {       this.buckets.add( servers[i] );       if ( log.isDebugEnabled() )          log.debug( "++++ added " + servers[i] + " to server bucket" );       }    }    else {       this.buckets.add( servers[i] );    }       // create initial connections    for ( int j = 0; j < initConn; j++ ) {            SockIO socket = createSocket( servers[i] );        if ( socket == null ) {          break;            }            addSocketToPool( availPool, servers[i], socket );    }  }   假如我们设置的servers是 192.168.0.1:44444和192.168.0.2:22222  然后我们设置了weights是  5和3 那么  buckets list的值最终会是  [  192.168.0.1:44444,  192.168.0.1:44444,  192.168.0.1:44444,  192.168.0.1:44444,  192.168.0.1:44444,  192.168.0.2:22222,  192.168.0.2:22222.  192.168.0.2:22222.  ]  然后就开始根据initCon初始连接数按servers分别创建socket  那么究竟这个buckets做什么用呢?  在我们使用set存放对象时会调用  Java代码  SockIOPool.SockIO sock = pool.getSock( key, hashCode );   看一看pool.getSock的代码  Java代码  // get initial bucket  long bucket = getBucket( key, hashCode );  String server = ( this.hashingAlg == CONSISTENT_HASH )          ? consistentBuckets.get( bucket )          : buckets.get( (int)bucket );   其中有段代码是这样的,看看getBucket  Java代码  private long getBucket( String key, Integer hashCode ) {      long hc = getHash( key, hashCode );      if ( this.hashingAlg == CONSISTENT_HASH ) {          return findPointFor( hc );      }      else {           long bucket = hc % buckets.size();          if ( bucket < 0 ) bucket *= -1;              return bucket;      }  }  

    先不管key和hashCode,我们看到首先算出一个hc值后会直接做hc%buckets.size()实际上就是根据buckets的数量散列,最终值一定是buckets.size()范围里的一个值 然后最终server值就根据buckets.get( (int)bucket )得到,那么假如我们得到bucket是3,则参照上面buckets 里的值,得到 list.get(3)=192.168.0.1:44444 所以会根据weight设置的值的不同得到不同的server ,如果 weights设置10:1 那buckets里就是10个相同的server和另一个不同的,将来散列得到的server很大可能性是servers里设置的第一个server。 

     

    原文地址:http://bachmozart.javaeye.com/blog/211836

    最新回复(0)