缓存研究之Xmemcached

1 简介
Xmemcached是一个高性能的基于java nio的memcached客户端。在经过三个RC版本后，正式发布1.10-final版本。
xmemcached特性一览：
1、高性能
2、支持完整的memcached文本协议,二进制协议将在1.2版本实现。
3、支持JMX，可以通过MBean调整性能参数、动态添加/移除server、查看统计等。
4、支持客户端统计
5、支持memcached节点的动态增减。
6、支持memcached分布:余数分布和一致性哈希分布。
7、更多的性能调整选项。

2 与Spring整合
XMemcached从1.1.2开始，能灵活方便的与Spring Framework整合在一起使用。
2.1 最简单例子

<bean name="memcachedClient" class="net.rubyeye.xmemcached.utils.XMemcachedClientFactoryBean"> 
<property name="servers"> 
<value>host1:port1 host2:port2</value>
</property>
</bean>

 

然后在bean中就可以使用memcachedClient了。
2.2 复杂一点儿的例子

<bean name="memcachedClient"
                class="net.rubyeye.xmemcached.utils.XMemcachedClientFactoryBean">
                <property name="servers">
                        <value>host1:port1 host2:port2 host3:port3</value>
                </property>
                <property name="weights">
                        <list>
                                <value>1</value>
                                <value>2</value>
                                <value>3</value>
                        </list>
                </property>
                <property name="sessionLocator">
                        <bean class="net.rubyeye.xmemcached.impl.KetamaMemcachedSessionLocator"></bean>
                </property>
                <property name="transcoder">
                        <bean class="net.rubyeye.xmemcached.transcoders.SerializingTranscoder" />
                </property>
                <property name="bufferAllocator">
                        <bean class="net.rubyeye.xmemcached.buffer.SimpleBufferAllocator"></bean>
                </property>
        </bean>

 

其中各参数的意义：

 

参数	含义
servers	服务器列表，格式：ip:port
weights	主机映射：host1对应1号、host2对应2号..
sessionLocator	Session 分配器，有自带的，影响分布式
transcoder	通信编码方式
bufferAllocator	缓冲区分配器

       注：

       默认标准Hash, hash(key) mod server_count (余数分布)

 

MemcachedClientBuilder builder = new XMemcachedClientBuilder(                                        AddrUtil.getAddresses("server1:11211 server2:11211 server3:11211")); MemcachedClient mc = builder.build();

 

可以改为Consistent Hash(一致性哈希)：

 

MemcachedClientBuilder builder = new XMemcachedClientBuilder(                                        AddrUtil.getAddresses("server1:11211 server2:11211 server3:11211")); builder.setSessionLocator(new KetamaMemcachedSessionLocator()); MemcachedClient mc = builder.build();

使用实例
3.1介绍
目前支持所有的memcached 文本协议(二进协议正在开发中，预计在1.2中会出现) ，现包括了 get/gets、set、add、replace、delete、append、prepend、cas、multi get/gets、incr、decr、version、stats、flush_all等。
3.2 简单实例
MemcachedClient client;
   try {
    client = new XMemcachedClient("localhost",11211);//默认端口
    // store a value for one hour(synchronously).
    String someObject = "缓存这个一个小时可以吗？";
    client.set("key", 3600, someObject);

    // Retrieve a value.(synchronously).
    Object getSomeObject = client.get("key");
    // delete
    client.delete("key");
    System.out.println(getSomeObject.toString());
   } catch (TimeoutException e) {
    e.printStackTrace();
   } catch (InterruptedException e) {
    e.printStackTrace();
   } catch (MemcachedException e) {
    e.printStackTrace();
   } catch (IOException e) {
    e.printStackTrace();
   }
以上代码是没有经过Spring管理的，XMemcachedClient是一个常驻程序。所以可以考虑将它置入到Spring中，配置方式前面已讲过。
使用方式见下：
private MemcachedClient memcachedClient;

public void setMemcachedClient(MemcachedClient memcachedClient) {
   this.memcachedClient = memcachedClient;
}
注入后直接可使用。

3.3 监听服务器映射
Xmemcached源代码中XMemcachedClientMBean的setServerWeight(String server, int weight)方法可以将被监听服务器进行编号。具体使用方式可以用：
MemcachedClientBuilder builder = new    XMemcachedClientBuilder(AddrUtil.getAddresses("localhost:12000 localhost:12001"),new int[]{1,3});   //编号为1,3
MemcachedClient memcachedClient=builder.build();
3.4动态改变被监听服务器
Xmemcached支持动态的增加或者删除监听服务器，方式如下：
MemcachedClient client=new XMemcachedClient(AddrUtil.getAddresses("server1:11211 server2:11211"));  
//Add two new memcached nodes  
client.addServer("server3:11211 server4:11211");  
//Remove memcached servers  
client.removeServer("server1:11211 server2:11211");
3.5 使用CAS更新缓存
CASOperation需要先gets获取cas值，然后再调用cas方法更新，XMemcached提供了一个包装类可以帮你搞定这两步。以下是一个使用实例：
class CASThread extends Thread {
        static final class IncrmentOperation implements CASOperation<Integer> {
                /*
                 *Max repeat times.if repeat times is great than this value,
                 *xmemcached will throw a TimeoutException.
                 */
                @Override
                public int getMaxTries() {
                        return Integer.MAX_VALUE;
                }

                //increase current value               
                @Override
                public Integer getNewValue(long currentCAS, Integer currentValue) {
                        return currentValue + 1; // 当前值+1
                }
        }

        private XMemcachedClient mc;
        private CountDownLatch cd;

        public CASThread(XMemcachedClient mc, CountDownLatch cdl) {
                super();
                this.mc = mc;
                this.cd = cdl;

        }

        public void run() {
                try {
                        //do the cas operation
                        if (mc.cas("a", 0, new IncrmentOperation()))
                                this.cd.countDown();
                } catch (Exception e) {
                        e.printStackTrace();
                }
        }
}

public class CASTest {

        public static void main(String[] args) throws Exception {
                if (args.length < 2) {
                        System.err.println("Usage:java CASTest [threadNum] [server]");
                    System.exit(1);
                }
                //threads num
                int NUM = Integer.parseInt(args[0]);
                XMemcachedClient mc = new XMemcachedClient(AddrUtil.getAddresses(args[1]));
                //initial value is 0
                mc.set("a", 0, 0);
                CountDownLatch cdl = new CountDownLatch(NUM);
                long start = System.currentTimeMillis();
                //start NUM threads to increment the value
                for (int i = 0; i < NUM; i++)
                        new CASThread(mc, cdl).start();

                cdl.await();
                System.out.println("test cas,timed:"
                                + (System.currentTimeMillis() - start));
                System.out.println("result=" + mc.get("a"));
                mc.shutdown();
        }
}