一篇文全麵讀懂緩存與隊列

作者介紹

索寧，擅長Python開發、MySQL、前端等眾多技術領域，曾負責眾多企業安全架構解決方案，涉獵行業有媒體、出版社、航空運輸、醫療、軍隊、政府、教育等。

本文主題：

Memcached
Redis
RabbitMQ

一、Memcached

1、簡介、安裝、使用

Memcached 是一個高性能的分布式內存對象緩存係統，用於動態 Web 應用以減輕數據庫負載壓力。它通過在內存中緩存數據和對象來減少讀取數據庫的次數，從而提高動態、數據庫驅動網站的速度。Memcached 基於一個存儲鍵/值對的 hashmap。其守護進程（daemon ）是用 C寫的，但是客戶端可以用任何語言來編寫，並通過 memcached 協議與守護進程通信。

Memcached 內存管理機製：

Menceched 通過預分配指定的內存空間來存取數據，所有的數據都保存在 memcached 內置的內存中。

利用 Slab Allocation 機製來分配和管理內存。按照預先規定的大小，將分配的內存分割成特定長度的內存塊，再把尺寸相同的內存塊分成組，這些內存塊不會釋放，可以重複利用。

當存入的數據占滿內存空間時，Memcached 使用 LRU 算法自動刪除不是用的緩存數據，即重用過期數據的內存空間。Memcached 是為緩存係統設計的，因此沒有考慮數據的容災問題，和機器的內存一樣，重啟機器將會丟失，如果希望服務重啟數據依然能保留，那麼就需要 sina 網開發的 Memcachedb 持久性內存緩衝係統，當然還有常見的 NoSQL 服務如 Redis。

默認監聽端口：11211

Memcached 安裝

源碼安裝啟動 Memcached 快速部署文檔

源碼安裝 Memcached PHP 客戶端

Memcached 啟動

memcached -d -m 10 -u root -l 218.97.240.118 -p 12000 -c 256 -P /tmp/memcached.pid

參數說明：

-d 是啟動一個守護進程

-m 是分配給Memcache使用的內存數量，單位是MB

-u 是運行Memcache的用戶

-l 是監聽的服務器IP地址

-p 是設置Memcache監聽的端口,最好是1024以上的端口

-c 選項是最大運行的並發連接數，默認是1024，按照你服務器的負載量來設定

-P 是設置保存Memcache的pid文件

Memcached 命令

存儲命令: set/add/replace/append/prepend/cas

獲取命令: get/gets

其他命令: delete/stats..

Memcached 管理

Memcached memadmin php工具界麵化管理安裝部署文檔

# memadmin php 工具管理（memcadmin-1.0.12.tar.gz）

1、安裝memadmin php工具。

2、登陸memadmin php。

web方式訪問：https://IP地址/memadmin/

默認用戶名密碼都為admin。

2、Python 操作 Memcached

1）安裝 API 及基本操作

python 操作 Memcached 使用 Python-memcached 模塊

下載安裝：https://pypi.python.org/pypi/python-memcached

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

mc.set("foo", "bar")

ret = mc.get('foo')

print ret

2）天生支持集群

python-memcached 模塊原生支持集群操作，其原理本質是在內存維護一個主機列表，數字為權重，為3即出現3次，相對應的幾率大

mc = memcache.Client([

('192.168.1.5:12000', 3), # 數字為權重

('192.168.1.9:12000', 1),

], debug=True)

# 那麼在內存中主機列表為：

# host_list = ["192.168.1.5","192.168.1.5","192.168.1.5","192.168.1.9",]

那麼問題來了，集群情況下如何選擇服務器存儲呢？

如果要創建設置一個鍵值對（如：k1 = "v1"），那麼它的執行流程如下：

將 k1 轉換成一個數字
將數字和主機列表的長度求餘數，得到一個值 N（N 的範圍： 0 <= N < 列表長度）
在主機列表中根據第2步得到的值為索引獲取主機，例如：host_list[N]
連接將第3步中獲取的主機，將 k1 = "v1" 放置在該服務器的內存中

獲取值的話也一樣

源碼、將字符串轉換為數字

3）add

添加一個鍵值對，如果 key 已經存在，重複添加執行 add 則拋出異常

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

mc.add('k1', 'v1')

# mc.add('k1', 'v2') # 報錯，對已經存在的key重複添加，失敗！！！

4）replace

replace 修改某個 key 的值，如果 key 不存在，則異常

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

# 如果memcache中存在kkkk，則替換成功，否則一場

mc.replace('kkkk','999')

5） set 和 set_multi

set 設置一個鍵值對，如果 key 不存在，則創建
set_multi 設置多個鍵值對，如果 key 不存在，則創建

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

mc.set('name', 'nick')

mc.set_multi({'name': 'nick', 'age': '18'})

6） delete 和 delete_multi

delete 刪除指定的一個鍵值對
delete_multi 刪除指定的多個鍵值對

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

mc..delete('name', 'nick')

mc.delete_multi({'name': 'nick', 'age': '18'})

7） get 和 get_multi

get 獲取一個鍵值對
get_multi 獲取多個鍵值對

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

val = mc.get('name')

item_dict = mc.get_multi(["name", "age",])

8）append 和 prepend

append 修改指定key的值，在該值後麵追加內容
prepend 修改指定key的值，在該值前麵插入內容

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

# 原始值： k1 = "v1"

mc.append('k1', 'after')

# k1 = "v1after"

mc.prepend('k1', 'before')

# k1 = "beforev1after"

9） decr 和 incr

incr 自增，將 Memcached 中的某個值增加 N （ N 默認為1 ）
decr 自減，將 Memcached 中的某個值減少 N （ N 默認為1 ）

mport memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True)

mc.set('k1', '666')

mc.incr('k1')

# k1 = 667

mc.incr('k1', 10)

# k1 = 677

mc.decr('k1')

# k1 = 676

mc.decr('k1', 10)

# k1 = 666

10） gets 和 cas

這兩個方法就是傳說中的鎖。為了避免髒數據的產生而生：

import memcache

mc = memcache.Client(['192.168.1.5:12000'], debug=True, cache_cas=True)

v = mc.gets('product_count')

# 如果有人在gets之後和cas之前修改了product_count，那下麵的設置將會執行失敗，剖出異常

mc.cas('product_count', "899")

本質：每次執行 gets 時，就從 memcache 中獲取一個自增的數字，通過 cas 去修改 gets 到的值時，會攜帶之前獲取的自增值和 memcache 中的自增值進行比較，如果相等，則可以提交，如果不相等，那表示在 gets 和 cas 執行之間，又有其他人執行了 gets（獲取了緩衝的指定值），如此一來有可能出現非正常的數據，則不允許修改，並報錯。

二、Redis

1、簡介、安裝、使用、實例

Remote Dictionary Server（Redis）是一個基於 key-value 鍵值對的持久化數據庫存儲係統。Redis 和 Memcached 緩存服務很像，但它支持存儲的 value 類型相對更多，包括 string (字符串)、list (鏈表)、set (集合)、zset (sorted set --有序集合)和 hash（哈希類型）。這些數據類型都支持 push/pop、add/remove 及取交集並集和差集及更豐富的操作，而且這些操作都是原子性的。在此基礎上，Redis 支持各種不同方式的排序。與Memcached 一樣，為了保證效率，數據都是緩存在內存中。區別的是 Redis 會周期性的把更新的數據寫入磁盤或者把修改操作寫入追加的記錄文件，並且在此基礎上實現了 master-slave (主從)同步。

Redis 的出現，再一定程度上彌補了 Memcached 這類 key-value 內存換乘服務的不足，在部分場合可以對關係數據庫起到很好的補充作用。Redis 提供了 Python，Ruby，Erlang，PHP 客戶端，使用方便。

官方文檔：

https://www.Redis.io/documentation

https://www.Redis.cn/

Redis 安裝和使用實例

Redis 源碼快速安裝文檔

配置並啟動 Redis 服務

客戶端連接命令及命令測試

Redis 的 php 客戶端拓展安裝

Redis 主從同步

Redis 負載均衡

至於 Redis 的負載均衡，方案有很多：

LVS、keepalived、Twemproxy

有時間再補上吧...

Redis 持久化

Redis持久化方式有兩種：

（1）RDB：對內存中數據庫狀態進行快照；

（2）AOF：把每條寫命令都寫入文件，類似mysql的binlog日誌

RDB。

將Redis在內存中的數據庫狀態保存到磁盤裏麵，RDB文件是一個經過壓縮的二進製文件，通過該文件可以還原生成RDB文件時的數據庫狀態。

RDB的生成方式：

（1）執行命令手動生成

有兩個Redis命令可以用於生成RDB文件，一個是SAVE，另一個是BGSAVE
SAVE命令會阻塞Redis服務器進程，直到RDB文件創建完畢為止，在服務器進程阻塞期間，服務器不能處理任何命令請求；
BGSAVE命令會派生出一個子進程，然後由子進程負責創建RDB文件，服務器進程（父進程）繼續處理命令請求，創建RDB文件結束之前，客戶端發送的BGSAVE和SAVE命令會被服務器拒絕。

（2）通過配置自動生成

可以設置服務器配置的save選項，讓服務器每隔一段時間自動執行一次BGSAVE命令；
可以通過save選項設置多個保存條件，但隻要其中任意一個條件被滿足，服務器就會執行BGSAVE命令。

例如：

save 900 1

save 300 10

save 60 10000

那麼隻要滿足以下三個條件中的任意一個，BGSAVE命令就會被執行：

服務器在900秒之內，對數據庫進行了至少1次修改
服務器在300秒之內，對數據庫進行了至少10次修改
服務器在60秒之內，對數據庫進行了至少10000次修改

AOF

AOF持久化是通過保存Redis服務器所執行的寫命令來記錄數據庫狀態的。

AOF文件刷新的方式，有三種：

（1）appendfsync always - 每提交一個修改命令都調用fsync刷新到AOF文件，非常非常慢，但也非常安全；

（2）appendfsync everysec - 每秒鍾都調用fsync刷新到AOF文件，很快，但可能會丟失一秒以內的數據；

（3）appendfsync no - 依靠OS進行刷新，Redis不主動刷新AOF，這樣最快，但安全性就差。

默認並推薦每秒刷新，這樣在速度和安全上都做到了兼顧。

數據恢複

RDB方式：

RDB文件的載入工作是在服務器啟動時自動執行的，沒有專門用於載入RDB文件的命令，隻要Redis服務器在啟動時檢測到RDB文件存在，它就會自動載入RDB文件，服務器在載入RDB文件期間，會一直處於阻塞狀態，直到載入工作完成為止。

AOF方式

服務器在啟動時，通過載入和執行AOF文件中保存的命令來還原服務器關閉之前的數據庫狀態，具體過程：

（1）載入AOF文件

（2）創建模擬客戶端

（3）從AOF文件中讀取一條命令

（4）使用模擬客戶端執行命令

（5）循環讀取並執行命令，直到全部完成

如果同時啟用了RDB和AOF方式，AOF優先，啟動時隻加載AOF文件恢複數據。

2、Python 操作 Redis

python 安裝 Redis 模塊：

$ sudo pip install Redis

$ sudo easy_install Redis

$ sudo python setup.py install

詳見：

https://github.com/WoLpH/Redis-py

https://pypi.python.org/pypi/Redis

https://Redislabs.com/python-Redis

API 的使用

1）操作模式

redis-py 提供兩個類 Redis 和 StrictRedis 用於實現 Redis 的操作命令，StrictRedis 用於實現大部分官方的命令，並使用官方的語法和命令，Redis 是 StrictRedis 的子類，用於向後兼容舊版本的 redis-py。

import redis

r =redis.redis(host='192.168.1.5', port=6379)

r.set('foo', 'Bar')

print r.get('foo')

2）連接池

redis-py 使用 connection pool 來管理對一個 Redis server 的所有連接，避免每次建立、釋放連接帶來的額外開銷。默認每個 Redis 實例都會維護著一個自己的連接池。也可以覆蓋直接建立一個連接池，然後作為參數 Redis，這樣就可以實現多個 Redis 實例共享一個連接池資源。實現客戶端分片或有連接如何管理更細的顆粒控製。

pool = redis.ConnectionPool(host='192.168.1.5', port=6379)

r = redis.redis(connection_pool=pool)

r.set('foo', 'Bar')

print r.get('foo')

3）操作

分為五種數據類型，見下圖：

①String 操作，String 在內存中格式是一個 name 對應一個 value 來存儲

②Hash 操作，Redis 中 Hash 在內存中的存儲格式類似字典。

③List操作，Redis 中的 List 在在內存中按照一個 name 對應一個 List 來存儲，像變量對應一個列表。

④Set 操作，Set 集合就是不允許重複的列表。

⑤有序集合，在集合的基礎上，為每個元素排序；元素的排序需要根據另外一個值來進行比較，所以對於有序集合，每一個元素有兩個值：值和分數，分數是專門來做排序的。

4）管道

默認情況下，Redis-py 每次在執行請求時都會創建和斷開一次連接操作（連接池申請連接，歸還連接池），如果想要在一次請求中執行多個命令，則可以使用 pipline 實現一次請求執行多個命令，並且默認情況下 pipline 是原子性操作。

見以下實例：

import redis

pool = redis.ConnectionPool(host='10.211.55.4', port=6379)

r = redis.redis(connection_pool=pool)

# pipe = r.pipeline(transaction=False)

pipe = r.pipeline(transaction=True)

r.set('name', 'nick')

r.set('age', '18')

pipe.execute()

5）發布和訂閱

發布者：服務器

訂閱者：Dashboad 和數據處理

發布訂閱的 Demo 如下：

RedisHelper

訂閱者：

發布者：

三、RabbitMQ

1、簡介、安裝、使用

RabbitMQ 是一個在 AMQP 基礎上完成的，可複用的企業消息係統。他遵循 Mozilla Public License 開源協議。

MQ 全稱為 Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方式。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。消息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用於諸如遠程過程調用的技術。排隊指的是應用程序通過隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。

流程上生產者把消息放到隊列中去, 然後消費者從隊列中取出消息。

Producing，生產者, 產生消息的角色.
Exchange，交換器, 在得到生產者產生的消息後, 把消息放入隊列的角色.
Queue，隊列，消息暫時保存的地方.
Consuming，消費者，把消息從隊列中取出的角色.
消息 Message

RabbitMQ安裝

2、使用API操作RabbitMQ

基於隊列 Queue 實現生產者消費者模型：

View Code

RabbitMQ 實現

1、acknowledgment 消息不丟失

no-ack ＝ False，如果消費者由於某些情況宕了(its channel is closed, connection is closed, or TCP connection is lost)，那 RabbitMQ 會重新將該任務放入隊列中。

在實際應用中，可能會發生消費者收到Queue中的消息，但沒有處理完成就宕機（或出現其他意外）的情況，這種情況下就可能會導致消息丟失。為了避免這種情況發生，我們可以要求消費者在消費完消息後發送一個回執給RabbitMQ，RabbitMQ收到消息回執（Message acknowledgment）後才將該消息從Queue中移除；如果RabbitMQ沒有收到回執並檢測到消費者的RabbitMQ連接斷開，則RabbitMQ會將該消息發送給其他消費者（如果存在多個消費者）進行處理。這裏不存在timeout概念，一個消費者處理消息時間再長也不會導致該消息被發送給其他消費者，除非它的RabbitMQ連接斷開。

這裏會產生另外一個問題，如果我們的開發人員在處理完業務邏輯後，忘記發送回執給RabbitMQ，這將會導致嚴重的bug——Queue中堆積的消息會越來越多；消費者重啟後會重複消費這些消息並重複執行業務邏輯……

消費者

2、durable 消息不丟失

如果我們希望即使在RabbitMQ服務重啟的情況下，也不會丟失消息，我們可以將Queue與Message都設置為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ消息不會丟失。但依然解決不了小概率丟失事件的發生（比如RabbitMQ服務器已經接收到生產者的消息，但還沒來得及持久化該消息時RabbitMQ服務器就斷電了），如果我們需要對這種小概率事件也要管理起來，那麼我們要用到事務。由於這裏僅為RabbitMQ的簡單介紹，所以這裏將不講解RabbitMQ相關的事務。

需要改兩處地方：

生產者