由于工作上对多语言之间数据缓存一致性的需要，个人分析了libmemcached的实现，对底层一致性的实现有了一些了解，这里分享一些分析的一些过程，也给自己做个笔记。我使用是php，从php的扩展开始分析应该来说是最方便的。
这里从php扩展的Memcached::add方法入手去找到需要的内容。方法文件为php_memcached.c
add方法的php扩展代码如下（文件memcached-2.0.1/php_memcached.c）：
PHP_METHOD(Memcached, add)
{
php_memc_store_impl(INTERNAL_FUNCTION_PARAM_PASSTHRU, MEMC_OP_ADD, 0);
}
这里有一个实现的包装，再看 php_memc_store_impl, 这里省略一些代码直接看库的调用操作
case MEMC_OP_ADD:
if (!server_key) {
status = memcached_add(m_obj->memc, key, key_len, payload, payload_len, expiration, flags);
} else {
status = memcached_add_by_key(m_obj->memc, server_key, server_key_len, key,
key_len, payload, payload_len, expiration, flags);
}
break;
下面部分是addByKey的，上面是add的调用，这里的m_obj->memc 就memcached_st, m_obj是memc_obj结构指针
struct memc_obj {
memcached_st *memc;
zend_bool compression;
enum memcached_serializer serializer;
enum memcached_compression_type compression_type;
#if HAVE_MEMCACHED_SASL
zend_bool has_sasl_data;
#endif
} *obj;
然后进入memcached_add方法，在libmemcached/storage.cc文件中，不同版本的libmemcached可能有不一样，我这里是libmemcached1.0.4
memcached_return_t memcached_add(memcached_st *ptr,
const char *key, size_t key_length,
const char *value, size_t value_length,
time_t expiration,
uint32_t flags)
{
memcached_return_t rc;
LIBMEMCACHED_MEMCACHED_ADD_START();
rc= memcached_send(ptr, key, key_length,
key, key_length, value, value_length,
expiration, flags, 0, ADD_OP);
if (rc == MEMCACHED_NOTSTORED or rc == MEMCACHED_DATA_EXISTS)
{
memcached_set_error(*ptr, rc, MEMCACHED_AT);
}
LIBMEMCACHED_MEMCACHED_ADD_END();
return rc;
}
这里就是一个调用和一些错误检测，继续往下走进入memcached_send函数， 此函数除了检测参数和初始化数据外，完成了数据分布的计算和存储。数据分布使用memcached_generate_hash_with_redistribution(ptr, group_key, group_key_length)，得到根据当前环境配置下key对应的服务器的索引。函数代码如下：
uint32_t memcached_generate_hash_with_redistribution(memcached_st *ptr, const char *key, size_t key_length)
{
uint32_t hash= _generate_hash_wrapper(ptr, key, key_length);
_regen_for_auto_eject(ptr);
return dispatch_host(ptr, hash);
}
这里完成对key的hash，然后计算落在那个服务器，从_generate_hash_wrapper函数知道最终计算hash的函数，
static inline uint32_t _generate_hash_wrapper(const memcached_st *ptr, const char *key, size_t key_length)
{
WATCHPOINT_ASSERT(memcached_server_count(ptr));
if (memcached_server_count(ptr) == 1)
return 0;
if (ptr->flags.hash_with_namespace)
{
size_t temp_length= memcached_array_size(ptr->_namespace) + key_length;
char temp[MEMCACHED_MAX_KEY];
if (temp_length > MEMCACHED_MAX_KEY -1)
return 0;
strncpy(temp, memcached_array_string(ptr->_namespace), memcached_array_size(ptr->_namespace));
strncpy(temp + memcached_array_size(ptr->_namespace), key, key_length);
return generate_hash(ptr, temp, temp_length);
}
else
{
return generate_hash(ptr, key, key_length);
}
}
hash_with_namespace 这个是参数配置里面是否有附加的PREFIX_KEY，这里就是调用generate_hash得到key的hash，底层使用libhashkit/digest.cc的libhashkit_digest函数进行计算。
然后进入dispatch_host函数，里面是分布式hash处理的最终实现部分
static uint32_t dispatch_host(const memcached_st *ptr, uint32_t hash)
{
switch (ptr->distribution)
{
case MEMCACHED_DISTRIBUTION_CONSISTENT:
case MEMCACHED_DISTRIBUTION_CONSISTENT_WEIGHTED:
case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA:
case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY:
{
uint32_t num= ptr->ketama.continuum_points_counter;
WATCHPOINT_ASSERT(ptr->ketama.continuum);
memcached_continuum_item_st *begin, *end, *left, *right, *middle;
begin= left= ptr->ketama.continuum;
end= right= ptr->ketama.continuum + num;
while (left < right)
{
middle= left + (right - left) / 2;
if (middle->value < hash)
left= middle + 1;
else
right= middle;
}
if (right == end)
right= begin;
return right->index;
}
case MEMCACHED_DISTRIBUTION_MODULA:
return hash % memcached_server_count(ptr);
case MEMCACHED_DISTRIBUTION_RANDOM:
return (uint32_t) random() % memcached_server_count(ptr);
case MEMCACHED_DISTRIBUTION_VIRTUAL_BUCKET:
{
return memcached_virtual_bucket_get(ptr, hash);
}
default:
case MEMCACHED_DISTRIBUTION_CONSISTENT_MAX:
WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
return hash % memcached_server_count(ptr);
}
/* NOTREACHED */
}
第一段case部分是根据一些配置参数，计算得key落在那个server上的server_key，用到了二分查找。后面的case有hash取模方式，随机取模方式等方式实现取服务器索引。