Memcache学习笔记之 客户端调用

上一篇搭建好了服务端.现在开始所使用客户端了。

客户端版本有很多,有一个是xmemcache 但是我使用的是java_memcached-release_2.5.0.zip

下载地址。https://github.com/gwhalin/Memcached-Java-Client/downloads

先启动服务端

./bin/memcached -d -m 128 -u root -p 11211 -vv -c 512 -P /tmp/memcached.pid
这里我加上-vv 参数

运行下面程序。可以看到服务端缓存存取变化

客户端的代码调用

package com.cn.montya.memcached;

import com.danga.MemCached.MemCachedClient;
import com.danga.MemCached.SockIOPool;

public class MemcacheTest {
    private static final String POOL_NAME="testMum";

    /**
     * @param args
     * @throws InterruptedException 
     */
    public static void main(String[] args) throws InterruptedException {
        /**
         * MemCachedClient 主要提供一些get/set/delete 方法
         * 来操纵分布式服务器的缓存数据
         * memcached 存取值。主要是通过2次hash
         * 客户端可以把数据存储在多台memcached上。当查询数据时，客户端首先参考节点列表计算出key的哈希值（阶段一哈希），
         * 进而选中一个节点；客户端 将请求发送给选中的节点，然后memcached节点通过一个内部的哈希算法（阶段二哈希），查找真正的数据。
         */
        MemCachedClient client = initClient();
        int i=0;
        while(i<2){
          Thread.sleep(1000);
          i++;
           //所用一些算法 得到key的hash值 然后存到对应的server中，也可以设置超时时间
          client.set(i+"test", "hello"+i);
         }
         i=0;
         while(i<2){
             Thread.sleep(1000);
             i++;　　　　　　　　
             System.out.println((String)client.get(i+"test"));
         }
    }
    
    public static MemCachedClient initClient(){
        MemCachedClient client=null;
        //可以指定getTinstance名字,默认名为default
        //SockIOPool 是一个自平衡的连接池
        SockIOPool sock = SockIOPool.getInstance(POOL_NAME);
        //设置服务器集群,我这里是2台
        // String server[] = {"127.0.0.1:11211"，“192.168.100.100：11211”};
        String server[] = {"127.0.0.1:11211"};
        sock.setServers(server);
        //设置每个集群的权重
        //Integer weights[]={2,2};       
        Integer weights[]={2};
        sock.setWeights(weights);
        //设置测试化链接,最小,最大链接参数
        sock.setInitConn(10);
        sock.setMinConn(10);
        sock.setMinConn(200);
        /**
         * 集群用 设置池的故障转移的标志 
         * 如果一个server 发生故障
         * 连接池放返回另一台server的 socket
         * 非集群则设为false
         */
        sock.setFailover( false );
        sock.setMaxIdle( 1000 * 60 * 60 * 6 );  //设置threadPool最大的空闲时间     
        sock.setMaintSleep(100);//平衡thread sleep 时间 单位毫秒。
        //这是开启一个nagle 算法。改算法避免网络中充塞小封包，提高网络的利用率
        sock.setNagle( true );
        //Sets the socket timeout for reads
        //连接建立后对socket 的读取超时 
        sock.setSocketTO(2000); 
        // Sets the socket timeout for connect
        sock.setSocketConnectTO(0);
        sock.setAliveCheck(true);  
        //集群采用 一致性算法
        //sock.setHashingAlg(3);
        //初始化连接池
        sock.initialize();
        //new MemCachedClient 需要和SockIOPool参数一致。
        client = new MemCachedClient(POOL_NAME);
        return client;
    }

}

下面是服务端缓存存取变化

weights 参数说。

一般来说.如果一个server 的权重为4.那么buckets 会存放4个这个server。

同理 如果另一台server 的权重为3.那么buckets会存在3个这样的server

此时buckets.size()==7;

附带源码部分实现.

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++) {
                    buckets.add(servers[i]);
                }
            } else {
                buckets.add(servers[i]);
            }
                 ........
          }

默认是TCP协议.

protected final SchoonerSockIO createSocket(String host) {

        SchoonerSockIO socket = null;
        try {
            if (isTcp) {
                socket = new TCPSockIO(this, host, bufferSize, this.socketTO, this.socketConnectTO, this.nagle);
            } else {
                socket = new UDPSockIO(this, host, bufferSize, socketTO);
            }
        } catch (Exception ex) {
            log.error("++++ failed to get SockIO obj for: " + host);
            //log.error(ex.getMessage(), ex);
            socket = null;
        }

        return socket;
    }

连接池初始化

/**
     * Initializes the pool.
     */
    public void initialize() {
        initDeadLock.lock();
        try {

            // 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");
            }
            // pools
            socketPool = new HashMap<String, ConcurrentLinkedQueue<SchoonerSockIO>>(servers.length);
            poolCurrentConn = new HashMap<String, AtomicInteger>(servers.length);
            // only create up to maxCreate connections at once

            // initalize our internal hashing structures
            //如果所用集群最好使用一致性hash算法 sock.setHashingAlg(3); 
            if (this.hashingAlg == CONSISTENT_HASH)
                populateConsistentBuckets();
            else
                populateBuckets();

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

        } finally {
            initDeadLock.unlock();
        }

    }

为什么要所用一致性hash 可以看这篇blog

http://xok.la/2010/06/memcache_consistent_hashing.html