spymemcached深入分析
author：智深
version：0.7
日志：
QQ：2548921609(技术交流)
一、简介
spymemcached 是一个 memcache 的客户端， 使用 NIO 实现。
分析 spymemcached 需要了解 NIO，memcached使用，memcached协议，参考资料中列出了有用的资源连接。
NIO是New I/O的缩写，Java里边大家一般称为异步IO，实际上对应Linux系统编程中的事件驱动IO（event-driven IO），是对 epoll 的封装。其它的IO模型还包括同步，阻塞，非阻塞，多路复用（select，poll）。阻塞/非阻塞是 fd 的属性，同步会跟阻塞配合，这样的应用会一直 sleep，直到IO完成被内核唤醒；同步非阻塞的话，第一次读取时，如果没有数据，应用线程会立刻返回，但是应用需要确定以什么样的策略进行后面的系统调用，如果是简单的while循环会导致CPU 100%，复杂的类似自旋的策略增加了应用编程的难度，因此同步非阻塞很少使用。多路复用是Linux早期的一个进程监控多个fd的方式，性能比较低，每次调用涉及3次循环遍历，具体分析见 。event-driven IO，应用注册 感兴趣的socket IO事件（READ,WRITE），调用wait开始sleep，当条件成立时，如数据到达（可读），写缓冲区可用（可写），内核唤醒应用线程，应用线程根据得到的socket执行同步的调用读/写 数据。
二、协议简介
memcachded服务器和客户端之间采用 TCP 的方式通信，自定义了一套字节流的格式。本文分析的文本协议的构建和其它文本协议类似，mc里面分成命令行和数据块行，命令行里指明数据块的字节数目，命令行和数据块后都跟随\r\n。重要的一点是服务器在读取数据块时是根据命令行里指定的字节数目，因此数据块中含有\r或\n并不影响服务器读块操作。数据块后必须跟随\r\n。
存储命令发送
<command name> <key> <flags> <exptime> <bytes> [noreply]\r\n
cas <key> <flags> <exptime> <bytes> <cas unique> [noreply]\r\n
<data block>\r\n
command name = "set", "add", "replace", "append" or "prepend"
flags - 32位整数 server并不操作这个数据 get时返回给客户端
exptime - 过期时间，可以是unix时间戳或偏移量，偏移量的话最大为30*24*60*60， 超过这个值，服务器会认为是unix时间戳
bytes - 数据块字节的个数
响应
<data block>\r\n
STORED\r\n - 成功
NOT_STORED\r\n - add或replace命令没有满足条件
EXISTS\r\n - cas命令 表明item已经被修改
NOT_FOUND\r\n - cas命令 item不存在
获取命令
发送
get <key>*\r\n
gets <key>*\r\n
<key>* - 空格分割的一个或多个字符串
响应
VALUE <key> <flags> <bytes> [<cas unique>]\r\n
<data block>\r\n
VALUE <key> <flags> <bytes> [<cas unique>]\r\n
<data block>\r\n
END\r\n
本文以 get 操作为例；key = someKey value=abcdABC中文
以字节流的形式最终发送的数据
[103, 101, 116, 32, 115, 111, 109, 101, 75, 101, 121, 13, 10, 0]
103 101 116 - "get"
32 - "" 空格
115 111 109 101 75 101 121 - someKey
13 10 - \r\n
接收到的数据
VALUE someKey 0 13
61 62 63 64 41 42 43 E4 B8 AD E6 96 87\r\n
END\r\n
删除命令发送
delete <key> [noreply]\r\n
响应
DELETED\r\n - 成功删除
NOT_FOUND\r\n - 删除的条目不存在
其它命令详见参考资料 mc 协议
三、spymemcached中的重要对象
简介spy是mc的客户端，因此spy中所有对象需要基于它要完成的 功能 和 到mc服务器的通信协议来进行设计。最重要的MemcachedClient表示mc集群的client，应用中单例即可。spy中的每一个mc节点，用MemcachedNode表示，这个对象内部含有一个channel，网络连接到mc节点。要根据key的哈希值查找某个mc节点，spy中使用NodeLocator，默认locator是ArrayModNodeLocator，这个对象内部含有所有的MemcachedNode，spy使用的hash算法都在对象DefaultHashAlgorithm中，默认使用NATIVE_HASH，也就是String.hashCode()。locator和client中间还有一个对象，叫MemcachedConnection ，它表示到mc集群的连接，内部持有locator。clent内部持有MemcachedConnection(mconn)。spy使用NIO实现，因此有一个selector，这个对象存在于mconn中。要和服务器进行各种操作的通信，协议数据发送，数据解析，spy中抽象为Operation，文本协议的get操作最终实现为net.spy.memcached.protocol.ascii.GetOperationImpl。为了实现工作线程和IO线程之间的调度，spy抽象出了一个 GetFuture，内部持有一个OperationFuture。
TranscodeService执行字节数据和对象之间的转换，spy中实现方式为任务队列+线程池，这个对象的实例在client中。
对象详解SpyObject - spy中的基类 定义 Logger
MemcachedConnection - 表示到多台 mc 节点的连接
MemcachedConnection - 详细属性
shouldOptimize - 是否需要优化多个连续的get操作 --> gets 默认true
addedQueue - 用来记录排队到节点的操作
selector - 监控到多个 mc 服务器的读写事件
locator - 定位某个 mc 服务器
GetFuture - 前端线程和工作线程交互的对象
--> OperationFuture
ConnectionFactory - 创建 MemcachedConnection 实例；创建操作队列；创建 OperationFactory；制定 Hash 算法。
DefaultConnectionFactory - 默认连接工厂
DefaultHashAlgorithm - Hash算法的实现类
MemcachedNode - 定义到 单个memcached 服务器的连接
TCPMemcachedNodeImpl -
AsciiMemcachedNodeImpl -
BinaryMemcachedNodeImpl -
TCPMemcachedNodeImpl - 重要属性
socketAddress - 服务器地址
rbuf - 读缓冲区 默认大小 16384
wbuf - 写缓冲区 默认大小 16384
writeQ - 写队列
readQ - 读队列
inputQueue - 输入队列 memcachclient添加操作时先添加到 inputQueue中
opQueueMaxBlockTime - 操作的最大阻塞时间 默认10秒
reconnectAttempt - 重连尝试次数 volatile
channel - socket 通道
toWrite - 要向socket发送的字节数
optimizedOp - 优化后的Operation 实现类是OptimizedGetImpl
sk - channel注册到selector后的key
shouldAuth - 是否需要认证 默认 false
authLatch - 认证需要的Latch
reconnectBlocked -
defaultOpTimeout - 操作默认超时时间 默认值 2.5秒
continuousTimeout - 连续超时次数
opFact - 操作工厂
MemcachedClient - 重要属性
mconn - MemcachedConnection
opFact - 操作工厂
transcoder - 解码器
tcService - 解码线程池服务
connFactory - 连接工厂
Operation - 所有操作的基本接口
BaseOperationImpl
OperationImpl
BaseGetOpImpl - initialize 协议解析 构建缓冲区
GetOperationImpl
OperationFactory - 为协议构建操作 比如生成 GetOperation
BaseOperationFactory
AsciiOperationFactory - 文本协议的操作工厂 默认的操作工厂
BinaryOperationFactory - 二进制协议的操作工厂
OperationFactory - 根据 protocol handlers 构建操作
BaseOperationFactory
AsciiOperationFactory - 支持 ascii protocol
BinaryOperationFactory - 支持 binary operations
NodeLocator - 根据 key hash 值查找节点
ArrayModNodeLocator - hash 值和节点列表长度取模，作为下标，简单的数组查询
KetamaNodeLocator - Ketama一致性hash的实现
Transcoder - 对象和字节数组之间的转换接口
BaseSerializingTranscoder
SerializingTranscoder - 默认的transcoder
TranscodeService - 异步的解码服务，含有一个线程池
FailureMode - node失效的模式
Redistribute - 节点失效后移动到下一个有效的节点 默认模式
Retry - 重试失效节点 直至恢复
Cancel - 取消操作
四、整体流程
初始化客户端执行new MemcachedClient(new InetSocketAddress("192.168.56.101", 11211))。初始化 MemcachedClient，内部初始化MemcachedConnection，创建selector，注册channel到selector，启动IO线程。
线程模型初始化完成后，把监听mc节点事件的线程，也就是调用select的线程，称为IO线程；应用执行 c.get("someKey")，把应用所在的线程称为工作线程。工作线程通常由tomcat启动，负责创建操作，加入节点的操作队列，工作线程通常有多个；IO线程负责从队列中拿到操作，执行操作。
工作线程工作线程最终会调用asyncGet，方法内部会创建CountDownLatch(1), GetFuture，GetOperationImpl（持有一个内部类，工作线程执行完成后，最终会调用 latch.countDown()），选择mc节点，操作op初始化（生成写缓冲区），把op放入节点等待队列inputQueue中，同时会把当前节点放入mc连接(mconn)的addedQueue属性中，最后唤醒selector。最终工作线程在latch上等待（默认超时2.5秒）IO线程的执行结果。
IO线程IO线程被唤醒后
1、handleInputQueue()。移动Operation从inputQueue到writeQ中。对添加到addedQueue中的每一个MemcachedNode分别进行处理。这个函数会处理所有节点上的所有操作，全部发送到mc服务器（之前节点上就有写操作的才这么处理，否则只是注册写事件）。
2、循环过程中，如果当前node中没有写操作，则判断writeQ，readQ中有操作，在SK上注册读/写事件；如果有写操作，需要执行handleWrites函数。这个函数内部首先做的是、填充缓冲区fillWriteBuffer()：从writeQ中取出一个可写的操作（remove掉取消的和超时的），改变操作的状态为WRITING，把操作的数据复制到写缓冲区（写缓冲区默认16K，操作的字节数从十几字节到1M，这个地方有复杂的处理，后面会详细分析，现在只考虑简单情况），复制完成后把操作状态变为READING，从writeQ中remove当前操作，把操作add到readQ当中，这个地方会再去复制pending的操作；‚、发送写缓冲区的内容，全部发送完成后，会再次去填充缓冲区fillWriteBuffer()（比如说一个大的命令，一个缓冲区不够）。循环，直到所有的写操作都处理完。ƒ、判断writeQ，readQ是否有操作，更新sk注册的读写事件。get操作的话，现在已经注册了读事件。
3、selector.select()
4、数据到达时，执行handleIO(sk)，处理读事件；执行channel.read(rbuf)；执行readFromBuffer()，解析数据，读取到END\r\n将操作状态置为COMPLETE。
五、初始化详细流程
1、默认连接工厂为 DefaultConnectionFactory。接着创建TranscodeService（解码的线程池，默认线程最多为10），创建AsciiOperationFactory（支持ascii协议的操作工厂，负责生成各种操作，比如 GetOperationImpl），创建MemcachedConnection，设置操作超时时间（默认2.5秒）。
2、DefaultConnectionFactory创建MemcachedConnection详细过程：创建reconnectQueue，addedQueue，设置shouldOptimize为true，设置maxDelay为30秒，设置opFact，设置timeoutExceptionThreshold为1000（超过这个值，关闭到 mc node 的连接），打开 Selector，创建nodesToShutdown，设置bufSize为16384字节，创建到每个node的 MemcachedNode（默认是AsciiMemcachedNodeImpl，这一步创建SocketChannel，连接到mc节点，注册到selector，设置sk为刚注册得到的SelectionKey），最后启动 MemcachedConnection 线程，进入事件处理的循环代码
while(running) handleIO()。
六、核心流程代码
1、工作线程一切从工作线程调用 c.get("someKey") 方法开始
基本流程是：创建操作(Operation)，操作初始化，查找节点，把操作加入节点的等待队列，唤醒IO线程，然后工作线程在Future上等待IO线程的执行结果
// 默认等待2.5秒return asyncGet(key, tc).get(2500, TimeUnit.MILLISECONDS)
// 内部类GetOperation.Callback，是工作线程和IO线程交互的类// IO线程得到所有的操作响应数据后，调用gotData方法// IO线程接收到END\r\n后，调用receivedStatus和complete方法publicGetFuture asyncGet(final String key, final Transcoder tc) { final CountDownLatch latch = new CountDownLatch(1); final GetFuture rv = new GetFuture (latch, operationTimeout, key); Operation op = opFact.get(key, new GetOperation.Callback() { private Future val = null; public void receivedStatus(OperationStatus status) { rv.set(val, status); } public void gotData(String k, int flags, byte[] data) { val = tcService.decode(tc, new CachedData(flags, data, tc.getMaxSize())); } public void complete() { latch.countDown(); } }); rv.setOperation(op); mconn.enqueueOperation(key, op); return rv; // 最终会在rv上调用get方法}
// 向节点中队列中添加操作 1、查找节点 2、放入队列// 查找节点，根据key的hash值，对节点数取模protected void addOperation(final String key, final Operation o) { MemcachedNode placeIn = null; MemcachedNode primary = locator.getPrimary(key); if (primary.isActive() || failureMode == FailureMode.Retry) { placeIn = primary; } else if (failureMode == FailureMode.Cancel) { o.cancel(); } else { for (Iterator i = locator.getSequence(key); placeIn == null && i.hasNext();) { MemcachedNode n = i.next(); if (n.isActive()) { placeIn = n; } } if (placeIn == null) { placeIn = primary; } } if (placeIn != null) { addOperation(placeIn, o); } else { }} // 最重要的方法 protected void addOperation(final MemcachedNode node, final Operation o) { o.setHandlingNode(node); o.initialize(); // 操作初始化，生成要发送的字节流数据，放到缓冲区中 node.addOp(o); // 添加到节点的inputQueue中 addedQueue.offer(node); // 有操作的节点放入 addedQueue中 Selector s = selector.wakeup(); // 唤醒IO线程} 工作线程和IO线程之间传递的Future对象，结构如下 GetFuture ---> OperationFuture ---> latch ---> 表示依赖关系
// 最终工作线程在 OperationFuture的get方法上等待latchpublic T get(long duration, TimeUnit units) { if (!latch.await(duration, units)) { // 等待2.5秒 MemcachedConnection.opTimedOut(op); if (op != null) { op.timeOut(); //2.5秒后，操作没有执行完，设置超时(IO线程会判断，如果超时，就remove) } // throw exception } return objRef.get(); // objRef是一个原子引用，来保证对象的安全发布（线程安全）}// objRef引用的是一个TranscodeService.Task(本身是个FutureTask)对象，如果没有压缩和序列化的话，最终工作线程会调用tc.decode方法，得到返回值。 2、IO线程
IO线程的操作循环
处理输入队列，注册写事件；执行写操作，注册读事件；处理读操作，解析结果。public void run() { while (running) { handleIO(); }} public void handleIO() throws IOException { handleInputQueue(); int selected = selector.select(delay); Set selectedKeys = selector.selectedKeys(); if (selectedKeys.isEmpty() && !shutDown) { // some code } else { for (SelectionKey sk : selectedKeys) { handleIO(sk); } selectedKeys.clear(); }} handleInputQueue处理addedQueue中的所有节点，对每一个节点复制inputQueue中的操作到writeQ中。注册读写事件。private void handleInputQueue() { if (!addedQueue.isEmpty()) { Collection toAdd = new HashSet (); Collection todo = new HashSet (); MemcachedNode qaNode = null; while ((qaNode = addedQueue.poll()) != null) { todo.add(qaNode); } for (MemcachedNode qa : todo) { boolean readyForIO = false; if (qa.isActive()) { if (qa.getCurrentWriteOp() != null) { readyForIO = true; } } else { toAdd.add(qa); } qa.copyInputQueue(); if (readyForIO) { try { if (qa.getWbuf().hasRemaining()) { handleWrites(qa.getSk(), qa); } } catch (IOException e) { lostConnection(qa); } } qa.fixupOps(); } addedQueue.addAll(toAdd); }} spy中注册读写事件的函数readQ不为空注册读事件；writeQ不为空注册写事件；网络没有连接上注册连接事件。public final void fixupOps() { SelectionKey s = sk; if (s != null && s.isValid()) { int iops = getSelectionOps(); s.interestOps(iops); } else { }} public final int getSelectionOps() { int rv = 0; if (getChannel().isConnected()) { if (hasReadOp()) { rv |= SelectionKey.OP_READ; } if (toWrite > 0 || hasWriteOp()) { rv |= SelectionKey.OP_WRITE; } } else { rv = SelectionKey.OP_CONNECT; } return rv;} public final boolean hasReadOp() { return !readQ.isEmpty();}public final boolean hasWriteOp() { return !(optimizedOp == null && writeQ.isEmpty());} 3、handleWrites(SelectionKey sk, MemcachedNode qa)
我能够想到的一些场景，这个状态机代码必须处理的 ⑴ 当前队列中有1个操作，操作要发送的字节数目小于16K ⑵ 当前队列中有1个操作，操作要发送的字节数目大于16K(很大的set操作) ⑶ 当前队列中有多个操作，操作要发送的字节数目小于16K ⑷ 当前队列中有多个操作，操作要发送的字节数目大于16K ⑸ 任意一次写操作wrote为0 summary：处理节点中writeQ和inputQueue中的所有操作。每次循环会尽量填满发送缓冲区，然后将发送缓冲区的内容全部发送到网络上，循环往复，没有异常的情况下，直至发送完数据。操作中发送的内容只要放入到发送缓冲区后，就把操作加入到readQ(spy中根据writeQ和readQ中有没有数据，来注册读写事件)。 执行时机：IO线程在select上休眠，被工作线程唤醒后，处理输入队列，把操作复制到writeQ 中，注册写事件；再次调用select，返回后，就会调用handleWrites()，数据全部发送后，会注册读事件。处理输入队列时，如果wbuf还有东西没有发送，那么会在select调用前，调用handleWrites函数。
private void handleWrites(SelectionKey sk, MemcachedNode qa) throws IOException { qa.fillWriteBuffer(shouldOptimize); ---> boolean canWriteMore = qa.getBytesRemainingToWrite() > 0; while (canWriteMore) { int wrote = qa.writeSome(); ---> qa.fillWriteBuffer(shouldOptimize); canWriteMore = wrote > 0 && qa.getBytesRemainingToWrite() > 0; }} -- 发送数据；执行一次后，wbuf可能还有数据未写完public final int writeSome() throws IOException { int wrote = channel.write(wbuf); toWrite -= wrote; return wrote;} -- 填充缓冲区toWrite=0 表明 写缓冲区以前的内容已经全部写入到网络中，这样才会进行下一次的填充写缓冲区操作会尽量填满16K的缓冲区(单一操作数据量很大比如500K；或多个操作数据量500K)当一个操作中的数据完全写入缓冲区后，操作的状态变成READING，从writeQ中移除当前操作。public final void fillWriteBuffer(boolean shouldOptimize) { if (toWrite == 0 && readQ.remainingCapacity() > 0) { getWbuf().clear(); Operation o=getNextWritableOp(); ---> while(o != null && toWrite < getWbuf().capacity()) { synchronized(o) { ByteBuffer obuf = o.getBuffer(); int bytesToCopy = Math.min(getWbuf().remaining(), obuf.remaining()); byte[] b = new byte[bytesToCopy]; obuf.get(b); getWbuf().put(b); if (!o.getBuffer().hasRemaining()) { o.writeComplete(); transitionWriteItem(); preparePending(); -- copyInputQueue() if (shouldOptimize) { optimize(); } o=getNextWritableOp(); } toWrite += bytesToCopy; } } getWbuf().flip(); } else { }} -- 获取节点写队列中下一个可写的操作如果操作已经取消（前端线程等待超时，取消操作），或超时（IO线程没有来得及执行操作，操作超时），那么把操作从队列中移除，继续查找下一个操作。把可写的操作的状态从WRITE_QUEUED变成WRITING，同时把操作放入读队列中。private Operation getNextWritableOp() { Operation o = getCurrentWriteOp(); --->④ while (o != null && o.getState() == OperationState.WRITE_QUEUED) { synchronized(o) { if (o.isCancelled()) { Operation cancelledOp = removeCurrentWriteOp();--->⑤ } else if (o.isTimedOut(defaultOpTimeout)) { Operation timedOutOp = removeCurrentWriteOp(); } else { o.writing(); if (!(o instanceof TapAckOperationImpl)) { readQ.add(o); } return o; } o = getCurrentWriteOp(); } } return o;} ④ -- 拿到当前写操作(并不从队列中移除)public final Operation getCurrentWriteOp() { return optimizedOp == null ? writeQ.peek() : optimizedOp;}⑤ -- remove当前写操作public final Operation removeCurrentWriteOp() { Operation rv = optimizedOp; if (rv == null) { rv = writeQ.remove(); } else { optimizedOp = null; } return rv;} 4、handleReadshandleReads(SelectionKey sk, MemcachedNode qa)从网络中读取数据，放入rbuf。解析rbuf，得到结果；private void handleReads(SelectionKey sk, MemcachedNode qa) throws IOException { Operation currentOp = qa.getCurrentReadOp(); if (currentOp instanceof TapAckOperationImpl) { // no response qa.removeCurrentReadOp(); return; } ByteBuffer rbuf = qa.getRbuf(); final SocketChannel channel = qa.getChannel(); int read = channel.read(rbuf); if (read < 0) { // some code } while (read > 0) { // 从网络中读数据一直读到0为止 rbuf.flip(); while (rbuf.remaining() > 0) { // 只要缓冲区有数据 就去解析操作 synchronized(currentOp) { currentOp.readFromBuffer(rbuf); // 从rbuf中解析响应 if (currentOp.getState() == OperationState.COMPLETE) { Operation op = qa.removeCurrentReadOp(); // 操作解析成功，移除 } else if (currentOp.getState() == OperationState.RETRY) { ((VBucketAware) currentOp).addNotMyVbucketNode(currentOp.getHandlingNode()); Operation op = qa.removeCurrentReadOp(); retryOps.add(currentOp); } } currentOp=qa.getCurrentReadOp(); } rbuf.clear(); read = channel.read(rbuf); }} // 解析rbuf；readType有两种取值，LINE 和 DATA，用来区分正在操作的数据是命令行还是数据块。解析过程中，分别调用工作线程传入到操作中的回调对象的三个方法，分别是：receivedStatus，gotData，complete。public void readFromBuffer(ByteBuffer data) throws IOException { while (getState() != OperationState.COMPLETE && data.remaining() > 0) { if (readType == OperationReadType.DATA) { handleRead(data); } else { int offset = -1; for (int i = 0; data.remaining() > 0; i++) { byte b = data.get(); if (b == '\r') { foundCr = true; } else if (b == '\n') { assert foundCr : "got a \\n without a \\r"; offset = i; foundCr = false; break; } else { assert !foundCr : "got a \\r without a \\n"; byteBuffer.write(b); } } if (offset >= 0) { String line = new String(byteBuffer.toByteArray(), CHARSET); byteBuffer.reset(); OperationErrorType eType = classifyError(line); if (eType != null) { handleError(eType, line); } else { handleLine(line); // 取到完整的一行后 调用这个函数 解析行数据 } } } }} // 处理命令行和END行// 解析命令行时，分析各种参数 data为数据块的字节数public final void handleLine(String line) { if (line.equals("END")) { if (hasValue) { getCallback().receivedStatus(END); } else { getCallback().receivedStatus(NOT_FOUND); } transitionState(OperationState.COMPLETE); data = null; } else if (line.startsWith("VALUE ")) { String[] stuff = line.split(" "); currentKey = stuff[1]; currentFlags = Integer.parseInt(stuff[2]); data = new byte[Integer.parseInt(stuff[3])]; if (stuff.length > 4) { casValue = Long.parseLong(stuff[4]); } readOffset = 0; hasValue = true; setReadType(OperationReadType.DATA); } else if (line.equals("LOCK_ERROR")) { getCallback().receivedStatus(LOCK_ERROR); transitionState(OperationState.COMPLETE); } else { assert false : "Unknown line type: " + line; }} 5、那个著名的bug
JAVA NIO bug 会导致 CPU 100%
int selected = selector.select(delay);
Set<SelectionKey> selectedKeys = selector.selectedKeys();
if (selectedKeys.isEmpty() && !shutDown) {
if (++emptySelects > DOUBLE_CHECK_EMPTY) {
for (SelectionKey sk : selector.keys()) {
if (sk.readyOps() != 0) {
handleIO(sk);
} else {
lostConnection((MemcachedNode) sk.attachment());
}
DOUBLE_CHECK_EMPTY = 256,当连续的select返回为空时，++emptySelects，超过256，连接到当前mc节点的socket channel关闭，放入重连队列。
七、调试 spymemcached
调试 spymemcached IO线程的过程中，工作线程放入到节点队列的操作很容易超时，因此需要继承DefaultConnectionFactory 复写相关方法。
public class AstuteConnectionFactory extends DefaultConnectionFactory {@Overridepublic boolean shouldOptimize() {return false;}@Overridepublic long getOperationTimeout() {return 3000000L; // 3000S}} 八、参考资料
NIO：
memcached：
protocol：
