`

memcached客户端源码分析

阅读更多

memcached的Java客户端有好几种,http://code.google.com/p/memcached/wiki/Clients 罗列了以下几种

Html代码  收藏代码
  1. spymemcached  
  2.   
  3.     * http://www.couchbase.org/code/couchbase/java  
  4.           o An improved Java API maintained by Matt Ingenthron and others at Couchbase.  
  5.           o Aggressively optimised, ability to run async, supports binary protocol, support Membase and Couchbase features, etc. See site for details.  
  6.   
  7. Java memcached client  
  8.   
  9.     * http://www.whalin.com/memcached  
  10.           o A Java API is maintained by Greg Whalin from Meetup.com.  
  11.   
  12. More Java memcached clients  
  13.   
  14.     * http://code.google.com/p/javamemcachedclient  
  15.     * http://code.google.com/p/memcache-client-forjava  
  16.     * http://code.google.com/p/xmemcached  
  17.   
  18. Integrations  
  19.   
  20.     * http://code.google.com/p/simple-spring-memcached  
  21.     * http://code.google.com/p/memcached-session-manager   

 

 


我看的是第二个:Java memcached client源码,代码很简洁,一共只有9个类,最主要的有以下三个
MemcachedClient.java     客户端,负责提供外出程序接口,如get/set方法等等

SockIOPool.java          一个自平衡的连接池
NativeHandler.java       负责部分数据类型的序列化

 

它包含以下几个部分
1:key的服务端分布
2:数据序列化和压缩
3:连接池(连接方式和池的动态自动平衡)
4:failover和failback机制
5:和memcached服务器的通讯协议
 
关于这几个点,我从key的set/get说起,会贯穿上面列举的4个部分。这个文章写下来,本来是作为一个笔记,思维比较跳跃,可能不是很连贯,如有疑问,欢迎站内交流。这个client的代码

很简洁明了,我也没有加过多注释,只是理了一个脉络。

 

 


从客户端自带的测试代码开始

Java代码  收藏代码
  1. package com.meetup.memcached.test;  
  2. import com.meetup.memcached.*;  
  3. import org.apache.log4j.*;  
  4.   
  5. public class TestMemcached  {   
  6.     public static void main(String[] args) {  
  7.         BasicConfigurator.configure();  
  8.         String[] servers = { "127.0.0.1:12000"};  
  9.         SockIOPool pool = SockIOPool.getInstance();  
  10.         pool.setServers( servers );  
  11.         pool.setFailover( true );//故障转移  
  12.      pool.setInitConn( 10 ); //初始化连接为10  
  13.         pool.setMinConn( 5 );//最小连接为5  
  14.         pool.setMaxConn( 250 );//最大连接为250  
  15.         pool.setMaintSleep( 30 );//平衡线程休眠时间为30ms  
  16.         pool.setNagle( false );//Nagle标志为false  
  17.         pool.setSocketTO( 3000 );//响应超时时间为3000ms  
  18.         pool.setAliveCheck( true );//需要可用状态检查  
  19.      //初始化连接池,默认名称为"default"  
  20.         pool.initialize();  
  21.         //新建一个memcached客户端,如果没有给名字  
  22.      MemcachedClient mcc = new MemcachedClient();  
  23.   
  24.         // turn off most memcached client logging:  
  25.         com.meetup.memcached.Logger.getLogger( MemcachedClient.class.getName() ).setLevel( com.meetup.memcached.Logger.LEVEL_WARN );  
  26.   
  27.         for ( int i = 0; i < 10; i++ ) {  
  28.             boolean success = mcc.set( "" + i, "Hello!" );  
  29.             String result = (String)mcc.get( "" + i );  
  30.             System.out.println( String.format( "set( %d ): %s", i, success ) );  
  31.             System.out.println( String.format( "get( %d ): %s", i, result ) );  
  32.         }  
  33.   
  34.         System.out.println( "\n\t -- sleeping --\n" );  
  35.         try { Thread.sleep( 10000 ); } catch ( Exception ex ) { }  
  36.   
  37.         for ( int i = 0; i < 10; i++ ) {  
  38.             boolean success = mcc.set( "" + i, "Hello!" );  
  39.             String result = (String)mcc.get( "" + i );  
  40.             System.out.println( String.format( "set( %d ): %s", i, success ) );  
  41.             System.out.println( String.format( "get( %d ): %s", i, result ) );  
  42.         }  
  43.     }  
  44. }  

 


 
以上代码大概做了这几件事情:
初始化一个连接池
新建一个memcached客户端
set一个key/value
get一个key,并且打印出value
这是我们实际应用中很常见的场景。


连接池的创建和初始化 
        连接池SockIOPool是非常重要的部分,它的好坏直接决定了客户端的性能。SockIOPool用一个HashMap持有多个连接池对象,连接池以名称作为标识,默认为"default"。看看

SockIOPool的getInstance方法就知道了。

Java代码  收藏代码
  1. public static SockIOPool getInstance() {  
  2.       return getInstance("default");  
  3.   }  
  4.    
  5.   public static synchronized SockIOPool getInstance(String poolName) {  
  6.       if (pools.containsKey(poolName)) return pools.get(poolName);  
  7.   
  8.       SockIOPool pool = new SockIOPool();  
  9.       pools.put(poolName, pool);  
  10.   
  11.       return pool;  
  12.   }  

 

 


连接池实例化完成后,还需要初始化,看看pool.initialize()做了什么:

Java代码  收藏代码
  1.    
  2.   
  3. public void initialize() {  
  4.    //这里以自身作为同步锁,防止被多次初始化  
  5.    synchronized (this) {  
  6.    // 如果已经被初始化了则终止初始化过程  
  7.    if (initialized && (buckets != null || consistentBuckets != null) && (availPool != null)&& (busyPool != null)) {  
  8.       log.error("++++ trying to initialize an already initialized pool");  
  9.       return;  
  10.    }    
  11.     <span style="color: #ff0000;">  // 可用连接集合</span>  
  12.       availPool = new HashMap<String, Map<SockIO, Long>>(servers.length * initConn);  
  13.      //工作连接集合  
  14.      busyPool = new HashMap<String, Map<SockIO, Long>>(servers.length * initConn);    
  15.      // 不可用连接集合           
  16.      deadPool = new IdentityHashMap<SockIO, Integer>();  
  17.     hostDeadDur = new HashMap<String, Long>();  
  18.    hostDead = new HashMap<String, Date>();  
  19.    maxCreate = (poolMultiplier > minConn) ? minConn : minConn / poolMultiplier;   
  20.    if (log.isDebugEnabled()) {  
  21.       log.debug("++++ initializing pool with following settings:");  
  22.       log.debug("++++ initial size: " + initConn);  
  23.       log.debug("++++ min spare   : " + minConn);  
  24.       log.debug("++++ max spare   : " + maxConn);  
  25.    }  
  26.    if (servers == null || servers.length <= 0) {  
  27.       log.error("++++ trying to initialize with no servers");  
  28.       throw new IllegalStateException("++++ trying to initialize with no servers");  
  29.    }  
  30.    // initalize our internal hashing structures  
  31.    if (this.hashingAlg == CONSISTENT_HASH) populateConsistentBuckets();  
  32.    else populateBuckets();  
  33.    // mark pool as initialized  
  34.    this.initialized = true;  
  35.    // start maint thread  
  36.    if (this.maintSleep > 0this.startMaintThread();  
  37.   }  
  38. }  

 
 

 

 

连接池的关闭

很简单,只是重置清空相关参数而已

Java代码  收藏代码
  1. public void shutDown() {  
  2.         synchronized (this) {  
  3.             if (log.isDebugEnabled()) log.debug("++++ SockIOPool shutting down...");  
  4.   
  5.             if (maintThread != null && maintThread.isRunning()) {  
  6.                 // stop the main thread  
  7.                 stopMaintThread();  
  8.   
  9.                 // wait for the thread to finish  
  10.                 while (maintThread.isRunning()) {  
  11.                     if (log.isDebugEnabled()) log.debug("++++ waiting for main thread to finish run +++");  
  12.                     try {  
  13.                         Thread.sleep(500);  
  14.                     } catch (Exception ex) {  
  15.                     }  
  16.                 }  
  17.             }  
  18.   
  19.             if (log.isDebugEnabled()) log.debug("++++ closing all internal pools.");  
  20.             closePool(availPool);  
  21.             closePool(busyPool);  
  22.             availPool = null;  
  23.             busyPool = null;  
  24.             buckets = null;  
  25.             consistentBuckets = null;  
  26.             hostDeadDur = null;  
  27.             hostDead = null;  
  28.             maintThread = null;  
  29.             initialized = false;  
  30.             if (log.isDebugEnabled()) log.debug("++++ SockIOPool finished shutting down.");  
  31.         }  
  32.     }  

 

 

 

 

连接池的自动平衡
SockIOPool的initialize()方法最后有这么一行代码

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

 这是在初始化完成后,启动线程池平衡线程

  

Java代码  收藏代码
  1. protected void startMaintThread() {  
  2.       if (maintThread != null) {  
  3.           if (maintThread.isRunning()) {  
  4.               log.error("main thread already running");  
  5.           } else {  
  6.               maintThread.start();  
  7.           }  
  8.       } else {  
  9.           maintThread = new MaintThread(this);  
  10.           maintThread.setInterval(this.maintSleep);  
  11.           maintThread.start();  
  12.       }  
  13.   }  

  

 

MaintThread的run方法

Java代码  收藏代码
  1. public void run() {  
  2.    this.running = true;  
  3.    while (!this.stopThread) {  
  4.        try {  
  5.            Thread.sleep(interval);  
  6.            // if pool is initialized, then  
  7.            // run the maintenance method on itself  
  8.            if (pool.isInitialized()) pool.selfMaint();  
  9.        } catch (Exception e) {  
  10.            break;  
  11.        }  
  12.    }  
  13.    this.running = false;  

 
 其实最终的平衡方法是SockIOPool.selfMaint()

  

Java代码  收藏代码
  1. protected void selfMaint() {  
  2.         if (log.isDebugEnabled()) log.debug("++++ Starting self maintenance....");  
  3.   
  4.         // go through avail sockets and create sockets  
  5.         // as needed to maintain pool settings  
  6.         Map<String, Integer> needSockets = new HashMap<String, Integer>();  
  7.   
  8.         synchronized (this) {  
  9.             // 先统计每个服务器实例的可用连接是否小于最小可用连接数  
  10.         for (Iterator<String> i = availPool.keySet().iterator(); i.hasNext();) {  
  11.                 String host = i.next();  
  12.                 Map<SockIO, Long> sockets = availPool.get(host);  
  13.   
  14.                 if (log.isDebugEnabled()) log.debug("++++ Size of avail pool for host (" + host + ") = "  
  15.                                                     + sockets.size());  
  16.   
  17.                 // if pool is too small (n < minSpare)  
  18.                 if (sockets.size() < minConn) {  
  19.                     // need to create new sockets  
  20.                     int need = minConn - sockets.size();  
  21.                     needSockets.put(host, need);  
  22.                 }  
  23.             }  
  24.         }  
  25.   
  26.         // 如果小于最小可用连接数,则要新建增加可用连接  
  27.      Map<String, Set<SockIO>> newSockets = new HashMap<String, Set<SockIO>>();  
  28.   
  29.         for (String host : needSockets.keySet()) {  
  30.             Integer need = needSockets.get(host);  
  31.   
  32.             if (log.isDebugEnabled()) log.debug("++++ Need to create " + need + " new sockets for pool for host: "  
  33.                                                 + host);  
  34.   
  35.             Set<SockIO> newSock = new HashSet<SockIO>(need);  
  36.             for (int j = 0; j < need; j++) {  
  37.                 SockIO socket = createSocket(host);  
  38.                 if (socket == nullbreak;  
  39.                 newSock.add(socket);  
  40.             }  
  41.             newSockets.put(host, newSock);  
  42.         }  
  43.   
  44.         // synchronize to add and remove to/from avail pool  
  45.         // as well as clean up the busy pool (no point in releasing  
  46.         // lock here as should be quick to pool adjust and no  
  47.         // blocking ops here)  
  48.         synchronized (this) {  
  49.             //将新建的连接添加到可用连接集合里  
  50.        for (String host : newSockets.keySet()) {  
  51.                 Set<SockIO> sockets = newSockets.get(host);  
  52.                 for (SockIO socket : sockets)  
  53.                     addSocketToPool(availPool, host, socket);  
  54.             }  
  55.   
  56.             for (Iterator<String> i = availPool.keySet().iterator(); i.hasNext();) {  
  57.                 String host = i.next();  
  58.                 Map<SockIO, Long> sockets = availPool.get(host);  
  59.                 if (log.isDebugEnabled()) log.debug("++++ Size of avail pool for host (" + host + ") = "  
  60.                                                     + sockets.size());  
  61.                    
  62.                 //如果可用连接超过了最大连接数,则要关闭一些  
  63.           if (sockets.size() > maxConn) {  
  64.                     // need to close down some sockets  
  65.                     int diff = sockets.size() - maxConn;  
  66.                     int needToClose = (diff <= poolMultiplier) ? diff : (diff) / poolMultiplier;  
  67.   
  68.                     if (log.isDebugEnabled()) log.debug("++++ need to remove " + needToClose  
  69.                                                         + " spare sockets for pool for host: " + host);  
  70.   
  71.                     for (Iterator<SockIO> j = sockets.keySet().iterator(); j.hasNext();) {  
  72.                         if (needToClose <= 0break;  
  73.   
  74.                         // remove stale entries  
  75.                         SockIO socket = j.next();  
  76.                         long expire = sockets.get(socket).longValue();  
  77.   
  78.                         // 这里回收可用连接池的闲置连接,连接设置到可用连接池里时,expire设置为当前时间。如果 (expire + maxIdle) < System.currentTimeMillis()为true,则表  
  79. 明,该连接在可用连接池呆得太久了,需要回收  
  80.                if ((expire + maxIdle) < System.currentTimeMillis()) {  
  81.                             if (log.isDebugEnabled()) log.debug("+++ removing stale entry from pool as it is past its idle timeout and pool is over max spare");  
  82.   
  83.                             // remove from the availPool  
  84.                             deadPool.put(socket, ZERO);  
  85.                             j.remove();  
  86.                             needToClose--;  
  87.                         }  
  88.                     }  
  89.                 }  
  90.             }  
  91.   
  92.             //清理正在工作的连接集合  
  93.         for (Iterator<String> i = busyPool.keySet().iterator(); i.hasNext();) {  
  94.                 String host = i.next();  
  95.                 Map<SockIO, Long> sockets = busyPool.get(host);  
  96.                 if (log.isDebugEnabled()) log.debug("++++ Size of busy pool for host (" + host + ")  = "  
  97.                                                     + sockets.size());  
  98.                 // loop through all connections and check to see if we have any hung connections  
  99.                 for (Iterator<SockIO> j = sockets.keySet().iterator(); j.hasNext();) {  
  100.                     // remove stale entries  
  101.                     SockIO socket = j.next();  
  102.                     long hungTime = sockets.get(socket).longValue();  
  103.                     //如果工作时间超过maxBusyTime,则也要回收掉,超过maxBusyTime,可能是服务器响应时间过长  
  104.              if ((hungTime + maxBusyTime) < System.currentTimeMillis()) {  
  105.                         log.error("+++ removing potentially hung connection from busy pool ... socket in pool for "  
  106.                                   + (System.currentTimeMillis() - hungTime) + "ms");  
  107.   
  108.                         // remove from the busy pool  
  109.                         deadPool.put(socket, ZERO);  
  110.                         j.remove();  
  111.                     }  
  112.                 }  
  113.             }  
  114.         }  
  115.   
  116.         // 最后清理不可用连接集合  
  117.      Set<SockIO> toClose;  
  118.         synchronized (deadPool) {  
  119.             toClose = deadPool.keySet();  
  120.             deadPool = new IdentityHashMap<SockIO, Integer>();  
  121.         }  
  122.   
  123.         for (SockIO socket : toClose) {  
  124.             try {  
  125.                 socket.trueClose(false);  
  126.             } catch (Exception ex) {  
  127.                 log.error("++++ failed to close SockIO obj from deadPool");  
  128.                 log.error(ex.getMessage(), ex);  
  129.             }  
  130.   
  131.             socket = null;  
  132.         }  
  133.   
  134.         if (log.isDebugEnabled()) log.debug("+++ ending self maintenance.");  
  135.     }  

 
 

 

 

key的服务器端分布

初始化方法其实就是根据每个服务器的权重,建立一个服务器地址集合,如果选择了一致性哈希,则对服务器地址进行一致性哈希分布,一致性哈希算法比较简单,如果不了解的同学,可以

自行google一下,initialize() 方法里有这段代码:

//一致性哈希

Java代码  收藏代码
  1. if (this.hashingAlg == CONSISTENT_HASH){  
  2.   populateConsistentBuckets();  
  3. }else populateBuckets();  

 
 看看populateConsistentBuckets()方法

// 用一致性哈希算法将服务器分布在一个2的32次方的环里,服务器的分布位置<=servers.length*40*4

Java代码  收藏代码
  1. private void populateConsistentBuckets() {  
  2.     if (log.isDebugEnabled()) log.debug("++++ initializing internal hashing structure for consistent hashing");  
  3.   
  4.     // store buckets in tree map  
  5.     this.consistentBuckets = new TreeMap<Long, String>();  
  6.     MessageDigest md5 = MD5.get();  
  7.     if (this.totalWeight <= 0 && this.weights != null) {  
  8.         for (int i = 0; i < this.weights.length; i++)  
  9.             this.totalWeight += (this.weights[i] == null) ? 1 : this.weights[i];  
  10.     } else if (this.weights == null) {  
  11.         this.totalWeight = this.servers.length;  
  12.     }  
  13.   
  14.     for (int i = 0; i < servers.length; i++) {  
  15.        int thisWeight = 1;  
  16.        if (this.weights != null && this.weights[i] != null) thisWeight = this.weights[i];  
  17.   
  18.       //这个值永远小于40 * this.servers.length,因为thisWeight/totalWeight永远小于1  
Java代码  收藏代码
  1.  double factor = Math.floor(((double) (40 * this.servers.length * thisWeight)) / (doublethis.totalWeight);  
  2.   
  3.   //服务器的分布位置为factor*4,factor<=40*this.servers.length,所以服务器的分布位置& lt;=40*this.servers.length*4。  
  4. for (long j = 0; j < factor; j++) {  
  5.       //md5值的二进制数组为16位  
  6.    byte[] d = md5.digest((servers[i] + "-" + j).getBytes());  
  7.        //16位二进制数组每4位为一组,每组第4个值左移24位,第三个值左移16位,第二个值左移8位,第一个值不移位。进行或运算,得到一个小于2的32 次方的long值。  
  8.     for (int h = 0; h < 4; h++) {  
  9.             Long k = ((long) (d[3 + h * 4] & 0xFF) << 24) | ((long) (d[2 + h * 4] & 0xFF) << 16)  
  10.                          | ((long) (d[1 + h * 4] & 0xFF) << 8) | ((long) (d[0 + h * 4] & 0xFF));  
  11.                 consistentBuckets.put(k, servers[i]);  
  12.                 if (log.isDebugEnabled()) log.debug("++++ added " + servers[i] + " to server bucket");  
  13.          }  
  14.    }  
  15.   
  16.    // create initial connections  
  17.    if (log.isDebugEnabled()) log.debug("+++ creating initial connections (" + initConn + ") for host: "  
  18.                                             + servers[i]);  
  19.   
  20.    //创建连接  
  21.  for (int j = 0; j < initConn; j++) {  
  22.        SockIO socket = createSocket(servers[i]);  
  23.        if (socket == null) {  
  24.             log.error("++++ failed to create connection to: " + servers[i] + " -- only " + j + " created.");  
  25.              break;  
  26.        }  
  27.   
  28.        //添加到可用连接池  
  29.    addSocketToPool(availPool, servers[i], socket);  
  30.        if (log.isDebugEnabled()) log.debug("++++ created and added socket: " + socket.toString()  
  31.                                                 + " for host " + servers[i]);  
  32.    }  
  33. }  

    

 

   如果不是一致性哈希,则只是普通分布,很简单,只是根据权重将服务器地址放入buckets这个List里

Java代码  收藏代码
  1. private void populateBuckets() {  
  2.         if (log.isDebugEnabled()) log.debug("++++ initializing internal hashing structure for consistent hashing");  
  3.   
  4.         // store buckets in tree map  
  5.         this.buckets = new ArrayList<String>();  
  6.   
  7.         for (int i = 0; i < servers.length; i++) {  
  8.             if (this.weights != null && this.weights.length > i) {  
  9.                 for (int k = 0; k < this.weights[i].intValue(); k++) {  
  10.                     this.buckets.add(servers[i]);  
  11.                     if (log.isDebugEnabled()) log.debug("++++ added " + servers[i] + " to server bucket");  
  12.                 }  
  13.             } else {  
  14.                 this.buckets.add(servers[i]);  
  15.                 if (log.isDebugEnabled()) log.debug("++++ added " + servers[i] + " to server bucket");  
  16.             }  
  17.   
  18.             // create initial connections  
  19.             if (log.isDebugEnabled()) log.debug("+++ creating initial connections (" + initConn + ") for host: "  
  20.                                                 + servers[i]);  
  21.   
  22.             for (int j = 0; j < initConn; j++) {  
  23.                 SockIO socket = createSocket(servers[i]);  
  24.                 if (socket == null) {  
  25.                     log.error("++++ failed to create connection to: " + servers[i] + " -- only " + j + " created.");  
  26.                     break;  
  27.                 }  
  28.   
  29.                 //新建连接后,加入到可用连接集合里  
  30.            addSocketToPool(availPool, servers[i], socket);  
  31.                 if (log.isDebugEnabled()) log.debug("++++ created and added socket: " + socket.toString()  
  32.                                                     + " for host " + servers[i]);  
  33.             }  
  34.         }  
  35.     }  

 
 

 

 

如何创建socket连接

在上面的private void populateBuckets()方法里,createSocket(servers[i])是创建到服务器的连接,看看这个方法

Java代码  收藏代码
  1. protected SockIO createSocket(String host) {  
  2. SockIO socket = null;  
  3. //hostDeadLock是一个可重入锁,它的变量声明为  
  4.   
  5.   
  6. private final ReentrantLock             hostDeadLock    = new ReentrantLock();  
  7. hostDeadLock.lock();  
  8. try {  
  9.   //hostDead.containsKey(host)为true表示曾经连接过该服务器,但没有成功。  
  10.   //hostDead是一个HashMap,key为服务器地址,value为当时连接不成功的时间  
  11.   //hostDeadDur是一个HashMap,key为服务器地址,value为设置的重试间隔时间  
  12.   
  13.    if (failover && failback && hostDead.containsKey(host) && hostDeadDur.containsKey(host)) {  
  14.        Date store = hostDead.get(host);  
  15.        long expire = hostDeadDur.get(host).longValue();  
  16.   
  17.       if ((store.getTime() + expire) > System.currentTimeMillis()) return null;  
  18.    }  
  19.  } finally {  
  20.     hostDeadLock.unlock();  
  21.   }  
  22.   
  23.   
  24. try {  
  25.      socket = new SockIO(this, host, this.socketTO, this.socketConnectTO, this.nagle);  
  26.      if (!socket.isConnected()) {  
  27.                log.error("++++ failed to get SockIO obj for: " + host + " -- new socket is not connected");  
  28.                deadPool.put(socket, ZERO);  
  29.                socket = null;  
  30.      }  
  31.  } catch (Exception ex) {  
  32.            log.error("++++ failed to get SockIO obj for: " + host);  
  33.            log.error(ex.getMessage(), ex);  
  34.            socket = null;  
  35.  }  
  36.   
  37.   // if we failed to get socket, then mark  
  38.   // host dead for a duration which falls off  
  39.   hostDeadLock.lock();  
  40.   try {  
  41.          //到了这里,socket仍然为null,说明这个server悲剧了,无法和它创建连接,则要把该server丢到不可用的主机集合里  
  42.       if (socket == null) {  
  43.                Date now = new Date();  
  44.                hostDead.put(host, now);  
  45.   
  46.                //如果上次就不可用了,到期了仍然不可用,就要这次的不可用时间设为上次的2倍,否则初始时长为1000ms  
  47.                long expire = (hostDeadDur.containsKey(host)) ? (((Long) hostDeadDur.get(host)).longValue() * 2) : 1000;  
  48.   
  49.                if (expire > MAX_RETRY_DELAY) expire = MAX_RETRY_DELAY;  
  50.   
  51.                hostDeadDur.put(host, new Long(expire));  
  52.                if (log.isDebugEnabled()) log.debug("++++ ignoring dead host: " + host + " for " + expire + " ms");  
  53.   
  54.                // 既然这个host都不可用了,那与它的所有连接当然要从可用连接集合"availPool"里删除掉  
  55.          clearHostFromPool(availPool, host);  
  56.            } else {  
  57.                if (log.isDebugEnabled()) log.debug("++++ created socket (" + socket.toString() + ") for host: " + host);  
  58.                //连接创建成功,如果上次不成功,那么这次要把该host从不可用主机集合里删除掉  
  59.           if (hostDead.containsKey(host) || hostDeadDur.containsKey(host)) {  
  60.                    hostDead.remove(host);  
  61.                    hostDeadDur.remove(host);  
  62.                }  
  63.            }  
  64.        } finally {  
  65.            hostDeadLock.unlock();  
  66.        }  
  67.   
  68.        return socket;  
  69.    }  

  

 

   SockIO构造函数

Java代码  收藏代码
  1. public SockIO(SockIOPool pool, String host, int timeout, int connectTimeout, boolean noDelay)  
  2.                                                                                                throws IOException,  
  3.                                                                                                UnknownHostException {  
  4.       this.pool = pool;  
  5.       String[] ip = host.split(":");  
  6.       // get socket: default is to use non-blocking connect  
  7.       sock = getSocket(ip[0], Integer.parseInt(ip[1]), connectTimeout);  
  8.       if (timeout >= 0this.sock.setSoTimeout(timeout);  
  9.       // testing only  
  10.       sock.setTcpNoDelay(noDelay);  
  11.       // wrap streams  
  12.       in = new DataInputStream(new BufferedInputStream(sock.getInputStream()));  
  13.       out = new BufferedOutputStream(sock.getOutputStream());  
  14.       this.host = host;  
  15.   }  

 
 

    getSocket方法

Java代码  收藏代码
  1. protected static Socket getSocket(String host, int port, int timeout) throws IOException {  
  2.          SocketChannel sock = SocketChannel.open();  
  3.          sock.socket().connect(new InetSocketAddress(host, port), timeout);  
  4.          return sock.socket();  
  5.  }  

  可以看到,socket连接是用nio方式创建的。

 

 

 

 新建MemcachedClient 
MemcachedClient mcc = new MemcachedClient();新建了一个memcached客户端,看看构造函数,没作什么,只是设置参数而已。

Java代码  收藏代码
  1. /** 
  2.   * Creates a new instance of MemCachedClient. 
  3.   */  
  4.  public MemcachedClient() {  
  5.      init();  
  6.  }  
  7.   
  8.   
  9.  private void init() {  
  10.      this.sanitizeKeys       = true;  
  11.      this.primitiveAsString  = false;  
  12.      this.compressEnable     = true;  
  13.      this.compressThreshold  = COMPRESS_THRESH;  
  14.      this.defaultEncoding    = "UTF-8";  
  15.      this.poolName           = ( this.poolName == null ) ? "default" : this.poolName;  
  16.   
  17.      // get a pool instance to work with for the life of this instance  
  18.      this.pool               = SockIOPool.getInstance( poolName );  
  19.  }  

 

 

 

 

 

 set方法如何工作

到此memcached客户端初始化工作完成。再回到测试类TestMemcached,看看for循环里的

boolean success = mcc.set( ""  + i, "Hello!" );
 String result = (String)mcc.get( "" + i );
 初始化后,就可以set,get了。看看set是怎么工作的。

 

Java代码  收藏代码
  1. /** 
  2.      * Stores data on the server; only the key and the value are specified. 
  3.      * 
  4.      * @param key key to store data under 
  5.      * @param value value to store 
  6.      * @return true, if the data was successfully stored 
  7.      */  
  8.     public boolean set( String key, Object value ) {  
  9.         return set( "set", key, value, nullnull, primitiveAsString );  
  10.     }  
  11.    
  12.   
  13.     //这个set方法比较长   
  14.    private boolean set( String cmdname, String key, Object value, Date expiry, Integer hashCode, boolean asString ) {  
  15.         if ( cmdname == null || cmdname.trim().equals( "" ) || key == null ) {  
  16.             log.error( "key is null or cmd is null/empty for set()" );  
  17.             return false;  
  18.         }  
  19.   
  20.         try {  
  21.             key = sanitizeKey( key );  
  22.         }  
  23.         catch ( UnsupportedEncodingException e ) {  
  24.             // if we have an errorHandler, use its hook  
  25.             if ( errorHandler != null )  
  26.                 errorHandler.handleErrorOnSet( this, e, key );  
  27.             log.error( "failed to sanitize your key!", e );  
  28.             return false;  
  29.         }  
  30.   
  31.         if ( value == null ) {  
  32.             log.error( "trying to store a null value to cache" );  
  33.             return false;  
  34.         }  
  35.   
  36.         // get SockIO obj  
  37.         SockIOPool.SockIO sock = pool.getSock( key, hashCode );  
  38.          
  39.         if ( sock == null ) {  
  40.             if ( errorHandler != null )  
  41.                 errorHandler.handleErrorOnSet( thisnew IOException( "no socket to server available" ), key );  
  42.             return false;  
  43.         }  
  44.          
  45.         if ( expiry == null )  
  46.             expiry = new Date(0);  
  47.   
  48.         // store flags  
  49.         int flags = 0;  
  50.          
  51.         // byte array to hold data  
  52.         byte[] val;  
  53.   
  54.     //这些类型自己序列化,否则由java序列化处理  
  55.    if ( NativeHandler.isHandled( value ) ) {           
  56.             if ( asString ) {  
  57.                 //如果是字符串,则直接getBytes    
  58.                 try {  
  59.                     if ( log.isInfoEnabled() )  
  60.                         log.info( "++++ storing data as a string for key: " + key + " for class: " + value.getClass().getName() );  
  61.                     val = value.toString().getBytes( defaultEncoding );  
  62.                 }  
  63.                 catch ( UnsupportedEncodingException ue ) {  
  64.                     // if we have an errorHandler, use its hook  
  65.                     if ( errorHandler != null )  
  66.                         errorHandler.handleErrorOnSet( this, ue, key );  
  67.                     log.error( "invalid encoding type used: " + defaultEncoding, ue );  
  68.                     sock.close();  
  69.                     sock = null;  
  70.                     return false;  
  71.                 }  
  72.             }  
  73.             else {  
  74.                 try {  
  75.                     if ( log.isInfoEnabled() )  
  76.                         log.info( "Storing with native handler..." );  
  77.                     flags |= NativeHandler.getMarkerFlag( value );  
  78.                     val    = NativeHandler.encode( value );  
  79.                 }  
  80.                 catch ( Exception e ) {  
  81.                     // if we have an errorHandler, use its hook  
  82.                     if ( errorHandler != null )  
  83.                         errorHandler.handleErrorOnSet( this, e, key );  
  84.                     log.error( "Failed to native handle obj", e );  
  85.   
  86.                     sock.close();  
  87.                     sock = null;  
  88.                     return false;  
  89.                 }  
  90.             }  
  91.         }  
  92.         else {  
  93.             // 否则用java的序列化  
  94.         try {  
  95.                 if ( log.isInfoEnabled() )  
  96.                     log.info( "++++ serializing for key: " + key + " for class: " + value.getClass().getName() );  
  97.                 ByteArrayOutputStream bos = new ByteArrayOutputStream();  
  98.                 (new ObjectOutputStream( bos )).writeObject( value );  
  99.                 val = bos.toByteArray();  
  100.                 flags |= F_SERIALIZED;  
  101.             }  
  102.             catch ( IOException e ) {  
  103.                 // if we have an errorHandler, use its hook  
  104.                 if ( errorHandler != null )  
  105.                     errorHandler.handleErrorOnSet( this, e, key );  
  106.   
  107.                 // if we fail to serialize, then  
  108.                 // we bail  
  109.                 log.error( "failed to serialize obj", e );  
  110.                 log.error( value.toString() );  
  111.   
  112.                 // return socket to pool and bail  
  113.                 sock.close();  
  114.                 sock = null;  
  115.                 return false;  
  116.             }  
  117.         }  
  118.          
  119.         //压缩内容  
  120.      if ( compressEnable && val.length > compressThreshold ) {  
  121.             try {  
  122.                 if ( log.isInfoEnabled() ) {  
  123.                     log.info( "++++ trying to compress data" );  
  124.                     log.info( "++++ size prior to compression: " + val.length );  
  125.                 }  
  126.                 ByteArrayOutputStream bos = new ByteArrayOutputStream( val.length );  
  127.                 GZIPOutputStream gos = new GZIPOutputStream( bos );  
  128.                 gos.write( val, 0, val.length );  
  129.                 gos.finish();  
  130.                 gos.close();  
  131.                  
  132.                 // store it and set compression flag  
  133.                 val = bos.toByteArray();  
  134.                 flags |= F_COMPRESSED;  
  135.   
  136.                 if ( log.isInfoEnabled() )  
  137.                     log.info( "++++ compression succeeded, size after: " + val.length );  
  138.             }  
  139.             catch ( IOException e ) {  
  140.                 // if we have an errorHandler, use its hook  
  141.                 if ( errorHandler != null )  
  142.                     errorHandler.handleErrorOnSet( this, e, key );  
  143.                 log.error( "IOException while compressing stream: " + e.getMessage() );  
  144.                 log.error( "storing data uncompressed" );  
  145.             }  
  146.         }  
  147.   
  148.         // now write the data to the cache server  
  149.         try {  
  150.              //按照memcached协议组装命令  
  151.         String cmd = String.format( "%s %s %d %d %d\r\n", cmdname, key, flags, (expiry.getTime() / 1000), val.length );  
  152.             sock.write( cmd.getBytes() );  
  153.             sock.write( val );  
  154.             sock.write( "\r\n".getBytes() );  
  155.             sock.flush();  
  156.   
  157.             // get result code  
  158.             String line = sock.readLine();  
  159.             if ( log.isInfoEnabled() )  
  160.                 log.info( "++++ memcache cmd (result code): " + cmd + " (" + line + ")" );  
  161.   
  162.             if ( STORED.equals( line ) ) {  
  163.                 if ( log.isInfoEnabled() )  
  164.                     log.info("++++ data successfully stored for key: " + key );  
  165.                 sock.close();  
  166.                 sock = null;  
  167.                 return true;  
  168.             }  
  169.             else if ( NOTSTORED.equals( line ) ) {  
  170.                 if ( log.isInfoEnabled() )  
  171.                     log.info( "++++ data not stored in cache for key: " + key );  
  172.             }  
  173.             else {  
  174.                 log.error( "++++ error storing data in cache for key: " + key + " -- length: " + val.length );  
  175.                 log.error( "++++ server response: " + line );  
  176.             }  
  177.         }  
  178.         catch ( IOException e ) {  
  179.   
  180.             // if we have an errorHandler, use its hook  
  181.             if ( errorHandler != null )  
  182.                 errorHandler.handleErrorOnSet( this, e, key );  
  183.   
  184.             // exception thrown  
  185.             log.error( "++++ exception thrown while writing bytes to server on set" );  
  186.             log.error( e.getMessage(), e );  
  187.   
  188.             try {  
  189.                 sock.trueClose();  
  190.             }  
  191.             catch ( IOException ioe ) {  
  192.                 log.error( "++++ failed to close socket : " + sock.toString() );  
  193.             }  
  194.   
  195.             sock = null;  
  196.         }  
  197.   
  198.         //用完了,就要回收哦,sock.close()不是真正的关闭,只是放入到可用连接集合里。    
  199.        if ( sock != null ) {  
  200.             sock.close();  
  201.             sock = null;  
  202.         }  
  203.         return false;  
  204.     }  

 通过set方法向服务器设置key和value,涉及到以下几个点
数据的压缩和序列化 (如果是get方法,则和set方法基本是相反的)
为key分配服务器
 对于一些常用类型,采用自定义的序列化,具体要看NativeHander.java,这个类比较简单,有兴趣可以自己看看

 

Java代码  收藏代码
  1. public static boolean isHandled( Object value ) {  
  2.        return (  
  3.            value instanceof Byte            ||  
  4.            value instanceof Boolean         ||  
  5.            value instanceof Integer         ||  
  6.            value instanceof Long            ||  
  7.            value instanceof Character       ||  
  8.            value instanceof String          ||  
  9.            value instanceof StringBuffer    ||  
  10.            value instanceof Float           ||  
  11.            value instanceof Short           ||  
  12.            value instanceof Double          ||  
  13.            value instanceof Date            ||  
  14.            value instanceof StringBuilder   ||  
  15.            value instanceof byte[]  
  16.            )  
  17.        ? true  
  18.        : false;  
  19.    }  

 其他类型则用java的默认序列化

 

 

 

为key选择服务器 
SockIOPool.SockIO sock = pool.getSock( key, hashCode );就是为key选择服务器

Java代码  收藏代码
  1. public SockIO getSock(String key, Integer hashCode) {  
  2.         if (log.isDebugEnabled()) log.debug("cache socket pick " + key + " " + hashCode);  
  3.         if (!this.initialized) {  
  4.             log.error("attempting to get SockIO from uninitialized pool!");  
  5.             return null;  
  6.         }  
  7.   
  8.         // if no servers return null  
  9.         if ((this.hashingAlg == CONSISTENT_HASH && consistentBuckets.size() == 0)  
  10.             || (buckets != null && buckets.size() == 0)) return null;  
  11.   
  12.         // if only one server, return it  
  13.         if ((this.hashingAlg == CONSISTENT_HASH && consistentBuckets.size() == 1)  
  14.             || (buckets != null && buckets.size() == 1)) {  
  15.             SockIO sock = (this.hashingAlg == CONSISTENT_HASH) ? getConnection(consistentBuckets.get(consistentBuckets.firstKey())) : getConnection(buckets.get(0));  
  16.             if (sock != null && sock.isConnected()) {  
  17.                 if (aliveCheck) {//健康状态检查  
  18.   
  19.                     if (!sock.isAlive()) {  
  20.                         sock.close();  
  21.                         try {  
  22.                             sock.trueClose();//有问题,真的关闭socket  
  23.   
  24.                         } catch (IOException ioe) {  
  25.                             log.error("failed to close dead socket");  
  26.                         }  
  27.                         sock = null;  
  28.                     }  
  29.                 }  
  30.             } else {//连接不正常,放入不可用连接集合里  
  31.           if (sock != null) {  
  32.                     deadPool.put(sock, ZERO);  
  33.                     sock = null;  
  34.                 }  
  35.             }  
  36.   
  37.             return sock;  
  38.         }  
  39.   
  40.         Set<String> tryServers = new HashSet<String>(Arrays.asList(servers));  
  41.         // get initial bucket  
  42.         long bucket = getBucket(key, hashCode);  
  43.         String server = (this.hashingAlg == CONSISTENT_HASH) ? consistentBuckets.get(bucket) : buckets.get((int) bucket);  
  44.         
  45.         while (!tryServers.isEmpty()) {  
  46.             // try to get socket from bucket  
  47.             SockIO sock = getConnection(server);  
  48.             if (log.isDebugEnabled()) log.debug("cache choose " + server + " for " + key);  
  49.             if (sock != null && sock.isConnected()) {  
  50.                 if (aliveCheck) {  
  51.                     if (sock.isAlive()) {  
  52.                         return sock;  
  53.                     } else {  
  54.                         sock.close();  
  55.                         try {  
  56.                             sock.trueClose();  
  57.                         } catch (IOException ioe) {  
  58.                             log.error("failed to close dead socket");  
  59.                         }  
  60.                         sock = null;  
  61.                     }  
  62.                 } else {  
  63.                     return sock;  
  64.                 }  
  65.             } else {  
  66.                 if (sock != null) {  
  67.                     deadPool.put(sock, ZERO);  
  68.                     sock = null;  
  69.                 }  
  70.             }  
  71.   
  72.             // if we do not want to failover, then bail here  
  73.             if (!failover) return null;  
  74.   
  75.             // log that we tried  
  76.             tryServers.remove(server);  
  77.   
  78.             if (tryServers.isEmpty()) break;   
  79.            //注意哦,下面是failover机制  
  80.         int rehashTries = 0;  
  81.             while (!tryServers.contains(server)) {  
  82.                 String newKey = String.format("%s%s", rehashTries, key);  
  83.                 if (log.isDebugEnabled()) log.debug("rehashing with: " + newKey);  
  84.   
  85.                 bucket = getBucket(newKey, null);  
  86.                 server = (this.hashingAlg == CONSISTENT_HASH) ? consistentBuckets.get(bucket) : buckets.get((int) bucket);  
  87.                 rehashTries++;  
  88.             }  
  89.         }  
  90.         return null;  
  91.     }  

 
 

 

   下面这个方法是真正的从服务器获取连接

 

    
   

Java代码  收藏代码
  1. public SockIO getConnection(String host) {  
  2.         if (!this.initialized) {  
  3.             log.error("attempting to get SockIO from uninitialized pool!");  
  4.             return null;  
  5.         }  
  6.   
  7.         if (host == nullreturn null;  
  8.   
  9.         synchronized (this) {  
  10.             // if we have items in the pool  
  11.             // then we can return it  
  12.             if (availPool != null && !availPool.isEmpty()) {  
  13.                 // take first connected socket  
  14.                 Map<SockIO, Long> aSockets = availPool.get(host);  
  15.                 if (aSockets != null && !aSockets.isEmpty()) {  
  16.                     for (Iterator<SockIO> i = aSockets.keySet().iterator(); i.hasNext();) {  
  17.                         SockIO socket = i.next();  
  18.                         if (socket.isConnected()) {  
  19.                             if (log.isDebugEnabled()) log.debug("++++ moving socket for host (" + host  
  20.                                                                 + ") to busy pool ... socket: " + socket);  
  21.                             // remove from avail pool  
  22.                             i.remove();  
  23.                             // add to busy pool  
  24.                             addSocketToPool(busyPool, host, socket);  
  25.                             // return socket  
  26.                             return socket;  
  27.                         } else {  
  28.                             // add to deadpool for later reaping  
  29.                             deadPool.put(socket, ZERO);  
  30.                             // remove from avail pool  
  31.                             i.remove();  
  32.                         }  
  33.                     }  
  34.                 }  
  35.             }  
  36.         }  
  37.   
  38.         // create one socket -- let the maint thread take care of creating more  
  39.         SockIO socket = createSocket(host);  
  40.         if (socket != null) {  
  41.             synchronized (this) {  
  42.                 addSocketToPool(busyPool, host, socket);  
  43.             }  
  44.         }  
  45.         return socket;  
  46.     }  
  47.   
  48.    

  

 

 

 failover和failback

这两者都是发生在获取可用连接这个环节。

failover,如果为key选择的服务器不可用,则对key重新哈希选择下一个服务器,详见getSock方法的末尾。

failback,用一个hashmap存储连接失败的服务器和对应的失效持续时间,每次获取连接时,都探测是否到了重试时间。

分享到:
评论

相关推荐

    易语言源码易语言Memcached协议客户端模块源码.rar

    易语言是一种专为中国人设计的编程语言,它以简体中文作为编程语法,降低了编程的门槛,使得更多非计算机专业的人也能进行程序开发。...同时,源码分析也有助于提高编程技能,理解网络编程和分布式系统的设计思想。

    memcached java源码(performance分支)

    **Memcached Java源码分析——Performance分支** Memcached是一款高性能的分布式内存对象缓存系统,广泛应用于Web应用中,用于缓解数据库的负载。在Java环境中,我们常常使用Java客户端库来与Memcached服务器进行...

    memcached工具类源码

    - **Memcached客户端连接器**:初始化和管理到Memcached服务器的连接,如建立Socket连接,处理心跳机制,保持会话活跃。 - **Key-Value操作接口**:封装了基本的set、get、delete等操作,供应用层调用。 - **配置...

    ibatis memcached 整合 源码 文档

    源码分析部分,你可以深入研究`MyMemcachedCache`类的实现,理解如何使用Memcached客户端API进行数据的存取。此外,`spymemcached`库中的源码也能帮助你理解Memcached与Java的交互过程。 文档方面,`doc`目录下的...

    memcached源码

    **memcached源码分析** `memcached`是一个高性能、分布式内存对象缓存系统,用于在动态系统中减少数据库负载,提升应用性能。它通过在内存中缓存数据和对象来减少读取数据库的次数,从而提高了网站的响应速度。本文...

    最新版Memcached for windows + 源码

    4. **Memcached与源码分析** - 源码阅读可以帮助开发者理解Memcached的内存管理、数据结构、网络通信等核心部分,为自定义扩展或优化提供基础。 - Memcached的内存管理采用了slab分配器,将内存划分为不同的slabs...

    安卓仿微博客户端源码&php;后端源码

    一、安卓客户端源码分析 1. 用户界面(UI)设计:安卓客户端的界面设计通常基于Android Studio,使用XML布局文件构建各种视图组件,如TextView、EditText、ImageView等,实现类似微博的发帖、浏览、评论等功能。此...

    memcached全面剖析.pdf

    #### 使用Cache::Memcached客户端 - **连接memcached**: 首先需要创建一个`Cache::Memcached`对象实例,并指定memcached服务器的地址和端口。 - **保存数据**: 通过`set`方法保存键值对到缓存中,例如`$cache-&gt;set...

    memcached之客户端memadmin安装

    **标题解析:** "memcached之客户端memadmin安装" 这个标题涉及到的是关于`memcached`的一...通过上述分析,我们可以构建一个全面的`memcached`客户端`memadmin`的安装和使用教程,帮助用户更好地理解和掌握这一工具。

    memcached开源库源码分享

    **Memcached开源库源码分析** Memcached是一款高性能、分布式内存对象缓存系统,它广泛应用于Web应用中,用于减轻数据库的负载,提高数据访问速度。这个开源库提供了高效的键值存储服务,允许应用程序快速存取临时...

    Memcached缓存技术资料源码

    本文将深入探讨Memcached的基本概念、工作原理、安装配置、使用方法以及源码分析。 **一、Memcached基本概念** 1. **缓存**: 缓存是一种存储技术,用于临时存储频繁访问的数据,减少对后端慢速存储(如数据库)的...

    memcached1.4.5源代码

    **memcached1.4.5源代码分析** `memcached`是一种高性能、分布式内存对象缓存系统,用于在Web应用程序中存储数据,以减少数据库负载。它的设计目标是减轻数据库的负担,通过缓存经常访问的数据来提高网络应用的响应...

    c# asp.net memcached client 调用示例

    在本文中,我们将深入探讨如何在C# ASP.NET项目中使用Memcached客户端进行缓存操作。Memcached是一种高性能、分布式内存对象缓存系统,广泛应用于缓解数据库负载,提高Web应用程序的响应速度。以下是一个关于如何...

    易语言-Memcached协议客户端模块

    通过分析这些源码,我们可以深入理解如何在易语言环境中构建网络客户端,如何解析和构建Memcached协议的请求和响应,以及如何处理网络通信中的异常和错误。 这个模块可能包括以下几个关键部分: 1. 连接管理:建立...

    Spring memcached 源码

    Spring Memcached 是一个用于在Spring应用中集成Memcached缓存服务的框架。Memcached是一种分布式内存对象缓存系统,常用于提高网站数据读取...通过源码分析,我们可以了解其工作原理,并根据实际需求进行调整和优化。

    memCache源码java客户端

    **memCache源码分析——Java客户端** memCache是一款高性能、分布式的内存对象缓存系统,常用于减轻数据库的负载,提升应用性能。它的主要特点是基于内存存储,操作速度快,无持久化机制,适用于缓存非关键数据。在...

    缓存全面剖析.pdf及安装文件、缓存客户端、源码

    2. **源码分析**:帮助开发者理解Memcached的内部实现,包括其数据结构、网络通信机制、线程模型和内存管理等。 3. **Memcached命令行工具**:如`telnet`命令行操作,以及如何进行基本的添加、删除、查看缓存项。 ...

Global site tag (gtag.js) - Google Analytics