memcached 客户端代码 Java memcached client学习（一致性hash）

昨天写了一篇短文 描述了淘宝面试的一些题。 
今天下午 装了在ubuntu中装了 memcached ，装起来还是很简单的。 
主要是装一下。 

memcached: wget http://memcached.googlecode.com/files/memcached-1.4.4.tar.gz  /data/
libevent:wget  http://monkey.org/~provos/libevent-1.4.13-stable.tar.gz   /

自己分配了一个端口。那么我想还是从最简单的memcached  -- 
java memcached client源码，总的代码量还是很少的 
主要是如下两个类: 
MemcachedClient.java 
SockIOPool.java 

好 先看推荐的测试代码: 

/**
 * Copyright (c) 2008 Greg Whalin
 * All rights reserved.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the BSD license
 *
 * This library is distributed in the hope that it will be
 * useful, but WITHOUT ANY WARRANTY; without even the implied
 * warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
 * PURPOSE.
 *
 * You should have received a copy of the BSD License along with this
 * library.
 *
 * @author greg whalin <[e-mail]>
 */
package com.meetup.memcached.test;

import com.meetup.memcached.*;
import org.apache.log4j.*;

public class TestMemcached  {
    public static void main(String[] args) {
              // memcached should be running on port 11211 but NOT on 11212

        BasicConfigurator.configure();
        String[] servers = { "localhost:11211"};
        SockIOPool pool = SockIOPool.getInstance();
        pool.setServers( servers );
        pool.setFailover( true );
        pool.setInitConn( 10 );
        pool.setMinConn( 5 );
        pool.setMaxConn( 250 );
        pool.setMaintSleep( 30 );
          //这是开启一个nagle 算法。改算法避免网络中充塞小封包，提高网络的利用率
        pool.setNagle( false );
        pool.setSocketTO( 3000 );
        pool.setAliveCheck( true );
        pool.initialize();

        MemcachedClient mcc = new MemcachedClient();

        // turn off most memcached client logging:
        com.meetup.memcached.Logger.getLogger( MemcachedClient.class.getName() ).setLevel( com.meetup.memcached.Logger.LEVEL_WARN );

        for ( int i = 0; i < 10; i++ ) {
            boolean success = mcc.set( "" + i, "Hello!" );
            String result = (String)mcc.get( "" + i );
            System.out.println( String.format( "set( %d ): %s", i, success ) );
            System.out.println( String.format( "get( %d ): %s", i, result ) );
        }

}

其实 对于我来说 我很想明白的是连接池是如何配置的,关键在于 pool.initialize(); 这个方法如何初始化的。 

/**
     * Initializes the pool.
     */
    public void initialize() {

        synchronized( this ) {

            // check to see if already initialized
            if ( initialized
                    && ( buckets != null || consistentBuckets != null )
                    && ( availPool != null )
                    && ( busyPool != null ) ) {
                log.error( "++++ trying to initialize an already initialized pool" );
                return;
            }

            // pools
            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();

            // mark pool as initialized
            this.initialized = true;

            // start maint thread
            if ( this.maintSleep > 0 )
                this.startMaintThread();
        }
    }

如上代码流程如下： 
1  检测是否已经被初始化 
2  定义可用链接 ，繁忙链接池 
3  判断是否一致性hash算法 还是普通的算法 
4  定义一个后台线程 ，来维护 
好 ，首先来分析下一致性hash算法。 

从如下代码来分析 ： 

    if (
              this.hashingAlg == CONSISTENT_HASH )
populateConsistentBuckets();

/** 将server添加到一致性hash的2的32次 圆环  **/
    private void populateConsistentBuckets() {
        if ( log.isDebugEnabled() )
            log.debug( "++++ initializing internal hashing structure for consistent hashing" );

        // store buckets in tree map
        this.consistentBuckets = new TreeMap<Long,String>();

        MessageDigest md5 = MD5.get();
        //得到总的权重
        if ( this.totalWeight <= 0 && this.weights !=  null ) {
            for ( int i = 0; i < this.weights.length; i++ )
                this.totalWeight += ( this.weights[i] == null ) ? 1 : this.weights[i];
        }
        else if ( this.weights == null ) {
            this.totalWeight = this.servers.length;
        }


        for ( int i = 0; i < servers.length; i++ ) {
            //每台服务器的权重
            int thisWeight = 1;
            if ( this.weights != null && this.weights[i] != null ) {
                thisWeight = this.weights[i];
            }

            //有兴趣的朋友可以参考平衡Hash 算法的另一个指标是平衡性 (Balance) ，定义如下：    平衡性　平衡性是指哈希的结果能够尽可能分布到所有的缓冲中去，这样可以使得所有的缓冲空间都得到利用
            //了解决这种情况， consistent hashing 引入了“虚拟节点”的概念，它可以如下定义： “虚拟节点”（ virtual node ）是实际节点在 hash 空间的复制品（ replica ），一实际个节点对应了若干个“虚拟节点”，这个对应个数也成为“复制个数”，“虚拟节点”在 hash 空间中以 hash 值排列。
            double factor = Math.floor(((double)(40 * this.servers.length * thisWeight)) / (double)this.totalWeight);

            for ( long j = 0; j < factor; j++ ) { //加密规则类似 127.0.0.1_1
                byte[] d = md5.digest( ( servers[i] + "-" + j ).getBytes() ); //转化成16位的字节数组
                //16位二进制数组每4位为一组，每组第4个值左移24位，第三个值左移16位，第二个值左移8位，第一个值不移位。进行或运算，得到一个小于2的32 次方的long值
                for ( int h = 0 ; h < 4; h++ ) { //因为是16位
                    Long k =  //实际上每个字节进行了运算
                          ((long)(d[3+h*4]&0xFF) << 24)
                        | ((long)(d[2+h*4]&0xFF) << 16)
                        | ((long)(d[1+h*4]&0xFF) << 8)
                        | ((long)(d[0+h*4]&0xFF));

                    consistentBuckets.put( k, servers[i] );
                    if ( log.isDebugEnabled() )
                        log.debug( "++++ added " + servers[i] + " to server bucket" );
                }
            }



            // create initial connections
            if ( log.isDebugEnabled() )
                log.debug( "+++ creating initial connections (" + initConn + ") for host: " + servers[i] );

            //创建链接
            for ( int j = 0; j < initConn; j++ ) {
                SockIO socket = createSocket( servers[i] );
                if ( socket == null ) {
                    log.error( "++++ failed to create connection to: " + servers[i] + " -- only " + j + " created." );
                    break;
                }
                //加入socket到连接池 这里慢慢谈
                addSocketToPool( availPool, servers[i], socket );
                if ( log.isDebugEnabled() )
                    log.debug( "++++ created and added socket: " + socket.toString() + " for host " + servers[i] );
            }
        }

    }

好 比如说 我们调用了 如下代码: 

MemcachedClient mcc = new MemcachedClient();
mcc.set("6", 1);

这里key 如何定位到一台server呢？我先把一致性hash算法的定位方法说下。 

//得到定位server的Socket封装对象
SockIOPool.SockIO sock = pool.getSock( key, hashCode );

//计算出key对应的hash值(md5) ，然后
long bucket = getBucket( key, hashCode );

//得到大于hash的map，因为treemap已经排好序了。调用tailMap可以得到大于等于这个hash的对象 ，然后调用firstKey得到圆环上的hash值
SortedMap<Long,String> tmap =
            this.consistentBuckets.tailMap( hv );
        return ( tmap.isEmpty() ) ? this.consistentBuckets.firstKey() : tmap.firstKey();

明天开始继续看代码 ，还是要坚持。 
一致性hash的算法参考 
http://xok.la/2010/06/memcache_consistent_hashing.html