缓存算法之LRU与LFU

1. LRU算法
1.1 背景
目前尽量由于摩尔定律，但是在存储硬件方面始终存在着差异，并且这种差异是不在同一数量级别的区别，例如在容量方面，内存<<外存；而在硬件成本与访问效率方面，内存>>外存。而目前互联网服务平台存在的特点：a. 读多写少，快速ms级响应，因此需要把数据搁在内存上；b. 数据规模巨大，长尾效应,由于数据规模巨大，只能把全量数据搁在外存上。正是由于服务场景需求与存储硬件特征之间的本身矛盾，缓存及相应的淘汰算法由此产生了：

一个在线服务平台其读取数据的过程：总是优先去离CPU最近的地方内存中读取数据，当有限的内存容量空间读取命中为空被击穿时，则会去外存的数据库或文件系统中读取；而当有限的缓存空间里“人满为患”时，而又有新的热点成员需要加入时，自然需要一定的淘汰机制。本能的基础淘汰算法：最后访问时间最久的成员最先会被淘汰(LRU)。

1.2 基本原理

由于队列具有先进先出的操作特点，所以通常用队列实现LRU，按最后访问的时间先进先出。

a. 利用队列类型对象，记录最近操作的元素,总是放在队首，这样最久未操作的元素自然被相对移动到队尾；同时，当元素总数达到上限值时，优先移除与淘汰队尾元素。
b. 利用HashMap辅助对象，快速检索队列中存在的元素。
1.3 操作
a. 写入操作： 新建一个元素，把元素插入队列首，当元素总和达到上限值时，同时删除队尾元素。
b. 读取操作：利用map快速检索，队列相关联的元素。
1.4 实现源码
a. 自定义链表方式

// A simple LRU cache written in C++
// Hash map + doubly linked list
#include <iostream>
#include <vector>
#include <ext/hash_map>
using namespace std;
using namespace __gnu_cxx;

template <class K, class T>
struct Node{
K key;
T data;
Node *prev, *next;
};

template <class K, class T>
class LRUCache{
public:
LRUCache(size_t size){
entries_ = new Node<K,T>[size];
for(int i=0; i<size; ++i)// 存储可用结点的地址
free_entries_.push_back(entries_+i);
head_ = new Node<K,T>;
tail_ = new Node<K,T>;
head_->prev = NULL;
head_->next = tail_;
tail_->prev = head_;
tail_->next = NULL;
}
~LRUCache(){
delete head_;
delete tail_;
delete[] entries_;
}
void Put(K key, T data){
Node<K,T> *node = hashmap_[key];
if(node){ // node exists
detach(node);
node->data = data;
attach(node);
}
else
{
if(free_entries_.empty())
{// 可用结点为空，即cache已满
node = tail_->prev;
detach(node);
hashmap_.erase(node->key);
}
else{
node = free_entries_.back();
free_entries_.pop_back();
}
node->key = key;
node->data = data;
hashmap_[key] = node;
attach(node);
}
}

T Get(K key){
Node<K,T> *node = hashmap_[key];
if(node){
detach(node);
attach(node);
return node->data;
}
else{// 如果cache中没有，返回T的默认值。与hashmap行为一致
return T();
}
}
private:
// 分离结点
void detach(Node<K,T>* node){
node->prev->next = node->next;
node->next->prev = node->prev;
}
// 将结点插入头部
void attach(Node<K,T>* node){
node->prev = head_;
node->next = head_->next;
head_->next = node;
node->next->prev = node;
}
private:
hash_map<K, Node<K,T>* > hashmap_;
vector<Node<K,T>* > free_entries_; // 存储可用结点的地址
Node<K,T> *head_, *tail_;
Node<K,T> *entries_; // 双向链表中的结点
};

int main(){
hash_map<int, int> map;
map[9]= 999;
cout<<map[9]<<endl;
cout<<map[10]<<endl;
LRUCache<int, string> lru_cache(100);
lru_cache.Put(1, "one");
cout<<lru_cache.Get(1)<<endl;
if(lru_cache.Get(2) == "")
lru_cache.Put(2, "two");
cout<<lru_cache.Get(2);
return 0;
}

b. 采用stl::list类型实现方式

#include <iostream>
#include <list>
#include <ext/hash_map>
#include <stdint.h>
#include <string>

template<class T1, class T2>
class Lru
{
public:
typedef std::list<std::pair<T1, T2> > List;
typedef typename List::iterator iterator;
typedef __gnu_cxx::hash_map<T1, iterator> Map;

Lru()
{
size_ = 1000;
resize(size_);
}

~Lru()
{
clear();
}

void resize(int32_t size)
{
if (size > 0)
{
size_ = size;
}
}

T2* find(const T1& first)
{
typename Map::iterator i = index_.find(first);

if (i == index_.end())
{
return NULL;
}
else
{
typename List::iterator n = i->second;
list_.splice(list_.begin(), list_, n);
return &(list_.front().second);
}
}

void remove(const T1& first)
{
typename Map::iterator i = index_.find(first);
if (i != index_.end())
{
typename List::iterator n = i->second;
list_.erase(n);
index_.erase(i);
}
}

void insert(const T1& first, const T2& second)
{
typename Map::iterator i = index_.find(first);
if (i != index_.end())
{ // found
typename List::iterator n = i->second;
list_.splice(list_.begin(), list_, n);
index_.erase(n->first);
n->first = first;
n->second = second;
index_[first] = n;
}
else if (size() >= size_ )
{ // erase the last element
typename List::iterator n = list_.end();
--n; // the last element
list_.splice(list_.begin(), list_, n);
index_.erase(n->first);
n->first = first;
n->second = second;
index_[first] = n;
}
else
{
list_.push_front(std::make_pair(first, second));
typename List::iterator n = list_.begin();
index_[first] = n;
}
}

/// Random access to items
iterator begin()
{
return list_.begin();
}

iterator end()
{
return list_.end();
}

int size()
{
return index_.size();
}

// Clear cache
void clear()
{
index_.clear();
list_.clear();
}

private:
int32_t size_;
List list_;
Map index_;
};

int main(void)
{
std::cout << "hello world " << std::endl;
tfs::Lru<uint32_t, std::string> name_cache;
name_cache.insert(1, "one");
name_cache.insert(2, "two");

std::string* value = name_cache.find(1);
const char* v = value->c_str();
std::cout << "result of the key 1 is: " << *name_cache.find(1) << std::endl;

return 0;
}

2. LRFU缓存算法

2.1 前言
缓存算法有许多种，各种缓存算法的核心区别在于它们的淘汰机制。通常的淘汰机制：所有元素按某种量化的重要程度进行排序，分值最低者则会被优先淘汰。而重要程度的量化指标又通常可以参考了两个维度：最后被访问的时间和最近被访问的频率次数。依次例如如下：

1. LRU(Least Recently Used ),总是淘汰最后访问时间最久的元素。

这种算法存在着问题：可能由于一次冷数据的批量查询而误淘汰大量热点的数据。

2. LFU(Least Frequently Used ),总是淘汰最近访问频率最小的元素。

这种算法也存在明显的问题: a. 如果频率时间度量是1小时，则平均一天每个小时内的访问频率1000的热点数据可能会被2个小时的一段时间内的访问频率是1001的数据剔除掉；b. 最近新加入的数据总会易于被剔除掉，由于其起始的频率值低。本质上其“重要性”指标访问频率是指在多长的时间维度进行衡量？其难以标定，所以在业界很少单一直接使用。也由于两种算法的各自特点及缺点，所以通常在生产线上会根据业务场景联合LRU与LFU一起使用，称之为LRFU。

2.2 实现机制
正是由于LRU与LFU算法各具特点，所以在行业生产线上通常会联合两者使用。我们可以在LRU的实现基础上稍作衍生，能实现LRFU缓存机制，即可以采用队列分级的思想，例如oracle利用两个队列维护访问的数据元素，按被访问的频率的维度把元素分别搁在热端与冷端队列；而在同一个队列内，最后访问时间越久的元素会越被排在队列尾。

参考：
1. https://en.wikipedia.org/wiki/Least_frequently_used
2. http://code.taobao.org/p/tfs/src/
3. http://www.hawstein.com/posts/lru-cache-impl.html