本章內容：
- Memcached
- 簡介、安裝、使用
- Python 操作 Memcached
- 天生支持集群
- redis
- 簡介、安裝、使用、實例
- Python 操作 Redis
- String、Hash、List、Set、Sort Set 操作
- 管道
- 發布訂閱
- RabbitMQ
- 簡介、安裝、使用
- 使用 API 操作 RabbitMQ
- 消息不丟失
- 發布訂閱
- 關鍵字發送
- 模糊匹配
一、Memcached
1、簡介、安裝、使用
Memcached 是一個高性能的分布式內存對象緩存系統，用於動態 Web 應用以減輕數據庫負載壓力。它通過在內存中緩存數據和對象來減少讀取數據庫的次數，從而提高動態、數據庫驅動網站的速度。Memcached 基於一個存儲鍵/值對的 hashmap。其守護進程（daemon ）是用 C 寫的，但是客戶端可以用任何語言來編寫，並通過 memcached 協議與守護進程通信。
Memcached 內存管理機制：
Menceched 通過預分配指定的內存空間來存取數據，所有的數據都保存在 memcached 內置的內存中。
利用 Slab Allocation 機制來分配和管理內存。按照預先規定的大小，將分配的內存分割成特定長度的內存塊，再把尺寸相同的內存塊分成組，這些內存塊不會釋放，可以重復利用。
當存入的數據占滿內存空間時，Memcached 使用 LRU 算法自動刪除不是用的緩存數據，即重用過期數據的內存空間。Memcached 是為緩存系統設計的，因此沒有考慮數據的容災問題，和機器的內存一樣，重啟機器將會丟失，如果希望服務重啟數據依然能保留，那么就需要 sina 網開發的 Memcachedb 持久性內存緩沖系統，當然還有常見的 NOSQL 服務如 redis。
默認監聽端口：11211
Memcached 安裝
wget http://memcached.org/latest tar -zxvf memcached-1.x.x.tar.gz cd memcached-1.x.x ./configure && make && make test && sudo make install PS：依賴libevent yum install libevent-devel apt-get install libevent-dev
# Memcached 服務安裝 # 1、安裝libevent mkdir /home/oldsuo/tools/ cd /home/oldsuo/tools/ wget http://down1.chinaunix.net/distfiles/libevent-2.0.21-stable.tar.gz ls libevent-2.0.21-stable.tar.gz tar zxf libevent-2.0.21-stable.tar.gz cd libevent-2.0.21-stable ./configure make && make install echo $? cd .. # 2、安裝Memcached wget http://memcached.org/files/memcached-1.4.24.tar.gz tar zxf memcached-1.4.24.tar.gz cd memcached-1.4.24 ./configure make make install echo $? cd .. # PS : memcached-1.4.24.tar -->客戶端 memcached-1.4.24.tar.gz -->服務端 # 3、啟動及關閉服務 echo "/usr/local/lib" >> /etc/ld.so.conf ldconfig # 查看幫助 /usr/local/bin/memcached –h # 啟動Memcached服務 memcached -p 11211 -u root -m 16m -c 10240 –d # 查看啟動狀態 lsof -i :11211 # 關閉服務 pkill memcached # memcached -p 11212 -u root -m 16m -c 10240 -d -P /var/run/11212.pid # kill `cat /var/run/11212.pid` # PS：開機自啟動把上述啟動命令放入/etc/rc.local
# Memcached PHP 客戶端安裝 cd /home/oldsuo/tools/ wget http://pecl.php.net/get/memcache-3.0.7.tgz tar zxf memcache-3.0.7.tgz cd memcache-3.0.7 /application/php/bin/phpize ./configure --enable-memcahce --with-php-config=/application/php/bin/php-config --with-zlib-dir make make install # 安裝完成后會有類似這樣的提示： Installing shared extensions: /application/php5.3.27/lib/php/extensions/no-debug-zts-20131226/ [root@localhost memcache-3.0.7]# ll /application/php5.3.27/lib/php/extensions/no-debug-zts-20131226/ total 1132 -rwxr-xr-x 1 root root 452913 Nov 17 16:52 memcache.so -rwxr-xr-x. 1 root root 157862 Oct 9 21:01 mysql.so -rwxr-xr-x. 1 root root 542460 Oct 9 19:25 opcache.so # 編輯php.ini文件，添加extension = memcache.so 一行 vim /application/php/lib/php.ini Extension_dir = "/application/php5.3.27/lib/php/extensions/no-debug-zts-20131226/" extension = memcache.so # 重啟 apache 服務是PHP的配置生效 [root@localhost application]# /usr/local/apache/bin/apachectl -t Syntax OK [root@localhost application]# /usr/local/apache/bin/apachectl graceful
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 管理
#1、telnet ip port 方式管理 telnet 127.0.0.1 11211 #2、命令直接操作，nc這樣的命令 [root@localhost application]# printf "stats slabs\r\n"|nc 127.0.0.1 11211 STAT active_slabs 0 STAT total_malloced 0 END #3、管理 Memcached 命令 a、stats 統計Memcached的各種信息。 b、stats reset 重新統計數據，重新開始統計。 c、stats slabs 顯示slabs信息。通過這命令能獲取每個slabs的chunksize長度，從而確定數據保存在哪個slab。 d、stats items 顯示slab中的item數目。 e、stats setting 查看一些Memcached設置，列如線程數…. f、stats slabs 查看slabs相關情況。 g、stats sizes 查看存在Item個數和大小。 h、stats cachedump 查看key value。 i、stats reset 清理統計數據。 j、set|get,gets 用來保存或獲取數據。
# memadmin php 工具管理（memcadmin-1.0.12.tar.gz） 1、安裝memadmin php工具。 cd /home/oldsuo/tools wget http://www.junopen.com/memadmin/memadmin-1.0.12.tar.gz tar zxf memadmin-1.0.12.tar.gz -C /usr/local/apache/htdocs/ ll /usr/local/apache/htdocs/memadmin/ 2、 登陸memadmin php。 web方式訪問：http://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" 放置在該服務器的內存中
獲取值的話也一樣
#!/usr/bin/env python #-*- coding:utf-8 -*- __author__ = 'Nick Suo' import binascii str_input = 'suoning' str_bytes = bytes(str_input, encoding='utf-8') num = (((binascii.crc32(str_bytes) & 0xffffffff) >> 16) & 0x7fff) or 1 print(num)
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 ）
import 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 客戶端，使用方便。
官方文檔：http://www.redis.io/documentation
Redis 安裝和使用實例
# Ubuntu 安裝 redis $ sudo apt-get install redis-server # 啟動服務端 $ sudo service redis-server {start|stop|restart|force-reload|status} # 啟動服務端 $ sudo redis-cli
# 源碼安裝 wget http://download.redis.io/releases/redis-3.0.6.tar.gz tar xzf redis-3.0.6.tar.gz cd redis-3.0.6 make # 啟動服務端 src/redis-server # 啟動客戶端 src/redis-cli
# 檢測后台進程是否存在 ps -ef |grep redis # 檢測6379端口是否在監聽 netstat -lntp | grep 6379 # 客戶端連接 $ sudo redis-cli 127.0.0.1:6379> set foo bar OK 127.0.0.1:6379> get foo "bar"
wget http://download.redis.io/releases/redis-3.0.5.tar.gz tar zxf redis-3.0.5.tar.gz cd redis-3.0.5 #less README make MALLOC=jemalloc make PREFIX=/application/redis-3.0.5 install -->指定安裝路徑 echo $? ln -s /application/redis-3.0.5/ /application/redis
[root@localhost redis-3.0.5]# tree /application/redis /application/redis `-- bin |-- redis-benchmark # Redis性能測試工具，測試Redis在系統及你的配置下的讀寫性能。 |-- redis-check-aof # 更新日志檢查。 |-- redis-check-dump # 用於本地數據庫檢查。 |-- redis-cli # Redis命令行操作工具。也可以telnet根據其純文本協議操作 |-- redis-sentinel -> redis-server `-- redis-server # Redis服務器的daemon啟動程序。 1 directory, 6 files
# 1、 配置環境變量 # 編輯vim /etc/profile添加一行 vim /etc/profile export PATH=/application/redis/bin/:$PATH tail -1 /etc/profile -->檢查 source /etc/profile -->生效 echo export PATH=/application/redis/bin/:$PATH >> /etc/profile tail -1 /etc/profile source /etc/profile # 2、 拷貝配置文件 [root@localhost redis-3.0.5]# pwd /home/oldSuo/tools/redis-3.0.5 -->解壓目錄 [root@localhost redis-3.0.5]# mkdir /application/redis/conf [root@localhost redis-3.0.5]# cp redis.conf /application/redis/conf/ cd /home/oldSuo/tools/redis-3.0.5 mkdir /application/redis/conf cp redis.conf /application/redis/conf/ # 3、 啟動redis redis-server /application/redis/conf/redis.conf & lsof -i :6379 [root@localhost redis-3.0.5]# lsof -i :6379 COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME redis-ser 5876 root 4u IPv6 793678202 0t0 TCP *:6379 (LISTEN) redis-ser 5876 root 5u IPv4 793678204 0t0 TCP *:6379 (LISTEN) # 4、 關閉redis redis-cli shutdown lsof -i :6379 -->檢查端口 #5、 啟動常見報錯 報錯：WARNING overcommit_memory is set to 0! Background save may fail under low memory condition. To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the command 'sysctl vm.overcommit_memory=1' for this to take effect. 解決： [root@localhost redis-3.0.5]# killall redis-server [root@localhost redis-3.0.5]# sysctl vm.overcommit_memory=1 vm.overcommit_memory = 1 永久生效：[root@localhost conf]# vim /etc/sysctl.conf 添加一行vm.overcommit_memory = 1
[root@localhost conf]# redis-cli --help [root@localhost conf]# redis-cli -h 192.168.200.95 [root@localhost conf]# redis-cli 127.0.0.1:6379> help redis-cli 3.0.5 Type: "help @<group>" to get a list of commands in <group> "help <command>" for help on <command> "help <tab>" to get a list of possible help topics "quit" to exit 127.0.0.1:6379> help get GET key summary: Get the value of a key since: 1.0.0 group: string 127.0.0.1:6379> help set SET key value [EX seconds] [PX milliseconds] [NX|XX] summary: Set the string value of a key since: 1.0.0 group: string 127.0.0.1:6379> set 007 oldSuo OK 127.0.0.1:6379> get 007 "oldSuo" 127.0.0.1:6379> 或者 [root@localhost conf]# redis-cli -h 192.168.200.95 -p 6379 set no005 suoning OK [root@localhost conf]# redis-cli -h 192.168.200.95 -p 6379 get no005 "suoning" 刪除並檢查 [root@localhost conf]# redis-cli del no005 (integer) 1 [root@localhost conf]# redis-cli get no005 (nil)
# 1、下載安裝 wget https://github.com/phpredis/phpredis/archive/master.zip unzip phpredis-master.zip cd phpredis-master /application/php/bin/phpize ./configure --with-php-config=/application/php/bin/php-config make make install [root@localhost phpredis-master]# make install Installing shared extensions: /application/php-5.6.8/lib/php/extensions/no-debug-non-zts-20131226/ [root@localhost phpredis-master]# cd /application/php-5.6.8/lib/php/extensions/no-debug-non-zts-20131226/ [root@localhost no-debug-non-zts-20131226]# ls memcache.so opcache.a opcache.so redis.so [root@localhost no-debug-non-zts-20131226]# # 2、修改php.ini設置，重啟php 在php.ini追加一條記錄 echo "extension = redis.so" >> /application/php/lib/php.ini #重啟 php-fpm killall php-fpm /application/php/sbin/php-fpm #網頁測試 ......
# 1、修改從庫redis.conf配置文件 #配置從庫redis.conf配置文件(先裝redis) #添加一行，主庫IP地址及端口 vim /application/redis/conf/redis.conf # slaveof <masterip> <masterport> slaveof 192.168.200.95 6379 # 2、重啟從庫redis服務 pkill redis redis-server /application/redis/conf/redis.conf & #啟動提示 7815:S 23 Nov 19:48:52.059 # Server started, Redis version 3.0.5 7815:S 23 Nov 19:48:52.060 * The server is now ready to accept connections on port 6379 7815:S 23 Nov 19:48:53.060 * Connecting to MASTER 192.168.200.95:6379 -->跟主庫建立連接 7815:S 23 Nov 19:48:53.060 * MASTER <-> SLAVE sync started -->主從同步已經開始 7815:S 23 Nov 19:48:53.062 * Non blocking connect for SYNC fired the event. 7815:S 23 Nov 19:48:53.074 * Master replied to PING, replication can continue... -->主從ping可以繼續 7815:S 23 Nov 19:48:53.075 * Partial resynchronization not possible (no cached master) 7815:S 23 Nov 19:48:53.087 * Full resync from master: 24b26f7abc62830a7ff97516c960ba7fc0992da9:1 7815:S 23 Nov 19:48:53.122 * MASTER <-> SLAVE sync: receiving 32 bytes from master -->接收到字節數 7815:S 23 Nov 19:48:53.122 * MASTER <-> SLAVE sync: Flushing old data 7815:S 23 Nov 19:48:53.122 * MASTER <-> SLAVE sync: Loading DB in memory 7815:S 23 Nov 19:48:53.122 * MASTER <-> SLAVE sync: Finished with success -->成功 # 3、測試主從同步 # 主庫：寫數據 [root@localhost redis]# redis-cli 127.0.0.1:6379> set test1 oldsuo OK # 從庫： [root@localhost conf]# redis-cli -h localhost -p 6379 monitor -->開啟實時監控 OK 1448280033.096372 [0 192.168.200.95:6379] "PING" 1448280043.125830 [0 192.168.200.95:6379] "PING" 1448280053.154134 [0 192.168.200.95:6379] "PING" 1448280070.858808 [0 192.168.200.95:6379] "SELECT" "0" 1448280070.858828 [0 192.168.200.95:6379] "set" "test1" "oldsuo" -->主庫添加數據，從庫同步 [root@localhost redis]# redis-cli -h 192.168.200.92 get test1 "oldsuo" -->從庫同步成功
至於 redis 的負載均衡，方案有很多：
LVS、keepalived、Twemproxy
小編有時間再補上吧...
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 or $ sudo easy_install redis or $ 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 來存儲
set(name, value, ex=None, px=None, nx=False, xx=False)
# 在Redis中設置值，默認，不存在則創建，存在則修改 # 參數： ex，過期時間（秒） px，過期時間（毫秒） nx，如果設置為True，則只有name不存在時，當前set操作才執行 xx，如果設置為True，則只有name存在時，崗前set操作才執行setnx(name, value)
# 設置值，只有name不存在時，執行設置操作（添加）setex(name, value, time)
# 設置值 # 參數： time，過期時間（數字秒 或 timedelta對象）psetex(name, time_ms, value)
# 設置值 # 參數： time_ms，過期時間（數字毫秒 或 timedelta對象）mset(*args, **kwargs)
# 批量設置值 # 如： mset(k1='v1', k2='v2') 或 mget({'k1': 'v1', 'k2': 'v2'})get(name)
# 獲取值mget(keys, *args)
# 批量獲取 # 如： mget('ylr', 'nick') 或 r.mget(['ylr', 'nick'])getset(name, value)
# 設置新值並獲取原來的值getrange(key, start, end)
# 獲取子序列（根據字節獲取，非字符） # 參數： name，Redis 的 name start，起始位置（字節） end，結束位置（字節） # 如： "索寧" ，0-3表示 "索"setrange(name, offset, value)
# 修改字符串內容，從指定字符串索引開始向后替換（新值太長時，則向后添加） # 參數： offset，字符串的索引，字節（一個漢字三個字節） value，要設置的值setbit(name, offset, value)
# 對name對應值的二進制表示的位進行操作 # 參數： # name，redis的name # offset，位的索引（將值變換成二進制后再進行索引） # value，值只能是 1 或 0 # 注：如果在Redis中有一個對應： n1 = "foo"， 那么字符串foo的二進制表示為：01100110 01101111 01101111 所以，如果執行 setbit('n1', 7, 1)，則就會將第7位設置為1， 那么最終二進制則變成 01100111 01101111 01101111，即："goo"getbit(name, offset)
# 獲取name對應的值的二進制表示中的某位的值 （0或1）bitcount(key, start=None, end=None)
# 獲取name對應的值的二進制表示中 1 的個數 # 參數： key，Redis的name start，位起始位置 end，位結束位置bitop(operation, dest, *keys)
# 獲取多個值，並將值做位運算，將最后的結果保存至新的name對應的值 # 參數： operation,AND（並） 、 OR（或） 、 NOT（非） 、 XOR（異或） dest, 新的Redis的name *keys,要查找的Redis的name # 如： bitop("AND", 'new_name', 'n1', 'n2', 'n3') 獲取Redis中n1,n2,n3對應的值，然后講所有的值做位運算（求並集），然后將結果保存 new_name 對應的值中strlen(name)
# 返回name對應值的字節長度（一個漢字3個字節）incr(self, name, amount=1)
# 自增 name對應的值，當name不存在時，則創建name＝amount，否則，則自增。 # 參數： name,Redis的name amount,自增數（必須是整數） # 注：同incrbincrbyfloat(self, name, amount=1.0)
# 自增 name對應的值，當name不存在時，則創建name＝amount，否則，則自增。 # 參數： name,Redis的name amount,自增數（浮點型）decr(self, name, amount=1)
# 自減 name對應的值，當name不存在時，則創建name＝amount，否則，則自減。 # 參數： name,Redis的name amount,自減數（整數）append(key, value)
# 在redis name對應的值后面追加內容 # 參數： key, redis的name value, 要追加的字符串
② Hash 操作，redis 中 Hash 在內存中的存儲格式類似字典
hset(name, key, value)
# name對應的hash中設置一個鍵值對（不存在，則創建；否則，修改） # 參數： name，redis的name key，name對應的hash中的key value，name對應的hash中的value # 注： hsetnx(name, key, value),當name對應的hash中不存在當前key時則創建（相當於添加）hmset(name, mapping)
# 在name對應的hash中批量設置鍵值對 # 參數： name，redis的name mapping，字典，如：{'k1':'v1', 'k2': 'v2'} # 如： # r.hmset('xx', {'k1':'v1', 'k2': 'v2'})hget(name,key)
# 在name對應的hash中獲取根據key獲取valuehmget(name, keys, *args)
# 在name對應的hash中獲取多個key的值 # 參數： name，reids對應的name keys，要獲取key集合，如：['k1', 'k2', 'k3'] *args，要獲取的key，如：k1,k2,k3 # 如： r.mget('xx', ['k1', 'k2']) 或 print r.hmget('xx', 'k1', 'k2')hgetall(name)
# 獲取name對應hash的所有鍵值hlen(name)
# 獲取name對應的hash中鍵值對的個數hkeys(name)
# 獲取name對應的hash中所有的key的值hvals(name)
# 獲取name對應的hash中所有的value的值hexists(name, key)
# 檢查name對應的hash是否存在當前傳入的keyhdel(name,*keys)
# 將name對應的hash中指定key的鍵值對刪除hincrby(name, key, amount=1)
自增name對應的hash中的指定key的值，不存在則創建key=amount 參數： name，redis中的name key， hash對應的key amount，自增數（整數）hincrbyfloat(name, key, amount=1.0)
# 自增name對應的hash中的指定key的值，不存在則創建key=amount # 參數： # name，redis中的name # key， hash對應的key # amount，自增數（浮點數） # 自增name對應的hash中的指定key的值，不存在則創建key=amounthscan(name, cursor=0, match=None, count=None)
# 增量式迭代獲取，對於數據大的數據非常有用，hscan可以實現分片的獲取數據，並非一次性將數據全部獲取完，從而放置內存被撐爆 # 參數： # name，redis的name # cursor，游標（基於游標分批取獲取數據） # match，匹配指定key，默認None 表示所有的key # count，每次分片最少獲取個數，默認None表示采用Redis的默認分片個數 # 如： # 第一次：cursor1, data1 = r.hscan('xx', cursor=0, match=None, count=None) # 第二次：cursor2, data1 = r.hscan('xx', cursor=cursor1, match=None, count=None) # ... # 直到返回值cursor的值為0時，表示數據已經通過分片獲取完畢hscan_iter(name, match=None, count=None)
# 利用yield封裝hscan創建生成器，實現分批去redis中獲取數據 # 參數： # match，匹配指定key，默認None 表示所有的key # count，每次分片最少獲取個數，默認None表示采用Redis的默認分片個數 # 如： # for item in r.hscan_iter('xx'): # print item
③ List操作，redis 中的 List 在在內存中按照一個 name 對應一個 List 來存儲，像變量對應一個列表。
lpush(name,values)
# 在name對應的list中添加元素，每個新的元素都添加到列表的最左邊 # 如： # r.lpush('oo', 11,22,33) # 保存順序為: 33,22,11 # 擴展： # rpush(name, values) 表示從右向左操作lpushx(name,value)
# 在name對應的list中添加元素，只有name已經存在時，值添加到列表的最左邊 # 更多： # rpushx(name, value) 表示從右向左操作llen(name)
# name對應的list元素的個數linsert(name, where, refvalue, value))
# 在name對應的列表的某一個值前或后插入一個新值 # 參數： # name，redis的name # where，BEFORE或AFTER # refvalue，標桿值，即：在它前后插入數據 # value，要插入的數據r.lset(name, index, value)
# 對name對應的list中的某一個索引位置重新賦值 # 參數： # name，redis的name # index，list的索引位置 # value，要設置的值r.lrem(name, value, num)
# 在name對應的list中刪除指定的值 # 參數： # name，redis的name # value，要刪除的值 # num， num=0，刪除列表中所有的指定值； # num=2,從前到后，刪除2個； # num=-2,從后向前，刪除2個lpop(name)
# 在name對應的列表的左側獲取第一個元素並在列表中移除，返回值則是第一個元素 # 更多： # rpop(name) 表示從右向左操作lindex(name, index)
# 在name對應的列表中根據索引獲取列表元素lrange(name, start, end)
# 在name對應的列表分片獲取數據 # 參數： # name，redis的name # start，索引的起始位置 # end，索引結束位置ltrim(name, start, end)
# 在name對應的列表中移除沒有在start-end索引之間的值 # 參數： # name，redis的name # start，索引的起始位置 # end，索引結束位置rpoplpush(src, dst)
# 從一個列表取出最右邊的元素，同時將其添加至另一個列表的最左邊 # 參數： # src，要取數據的列表的name # dst，要添加數據的列表的nameblpop(keys, timeout)
# 將多個列表排列，按照從左到右去pop對應列表的元素 # 參數： # keys，redis的name的集合 # timeout，超時時間，當元素所有列表的元素獲取完之后，阻塞等待列表內有數據的時間（秒）, 0 表示永遠阻塞 # 更多： # r.brpop(keys, timeout)，從右向左獲取數據brpoplpush(src, dst, timeout=0)
# 從一個列表的右側移除一個元素並將其添加到另一個列表的左側 # 參數： # src，取出並要移除元素的列表對應的name # dst，要插入元素的列表對應的name # timeout，當src對應的列表中沒有數據時，阻塞等待其有數據的超時時間（秒），0 表示永遠阻塞自定義增量迭代
# 由於redis類庫中沒有提供對列表元素的增量迭代，如果想要循環name對應的列表的所有元素，那么就需要： # 1、獲取name對應的所有列表 # 2、循環列表 # 但是，如果列表非常大，那么就有可能在第一步時就將程序的內容撐爆，所有有必要自定義一個增量迭代的功能： def list_iter(name): """ 自定義redis列表增量迭代 :param name: redis中的name，即：迭代name對應的列表 :return: yield 返回 列表元素 """ list_count = r.llen(name) for index in xrange(list_count): yield r.lindex(name, index) # 使用 for item in list_iter('pp'): print item
④ Set 操作，Set 集合就是不允許重復的列表
sadd(name,values)
# name對應的集合中添加元素scard(name)
# 獲取name對應的集合中元素個數sdiff(keys, *args)
# 在第一個name對應的集合中且不在其他name對應的集合的元素集合sdiffstore(dest, keys, *args)
# 獲取第一個name對應的集合中且不在其他name對應的集合，再將其新加入到dest對應的集合中sinter(keys, *args)
# 獲取多一個name對應集合的並集sinterstore(dest, keys, *args)
# 獲取多一個name對應集合的並集，再講其加入到dest對應的集合中sismember(name, value)
# 檢查value是否是name對應的集合的成員smembers(name)
# 獲取name對應的集合的所有成員smove(src, dst, value)
# 將某個成員從一個集合中移動到另外一個集合spop(name)
# 從集合的右側（尾部）移除一個成員，並將其返回srandmember(name, numbers)
# 從name對應的集合中隨機獲取 numbers 個元素srem(name, values)
# 在name對應的集合中刪除某些值sunion(keys, *args)
# 獲取多一個name對應的集合的並集sunionstore(dest,keys, *args)
# 獲取多一個name對應的集合的並集，並將結果保存到dest對應的集合中sscan(name, cursor=0, match=None, count=None)
sscan_iter(name, match=None, count=None)# 同字符串的操作，用於增量迭代分批獲取元素，避免內存消耗太大
⑤ 有序集合，在集合的基礎上，為每個元素排序；元素的排序需要根據另外一個值來進行比較，所以對於有序集合，每一個元素有兩個值：值和分數，分數是專門來做排序的。
zadd(name, *args, **kwargs)# 在name對應的有序集合中添加元素 # 如： # zadd('zz', 'n1', 1, 'n2', 2) # 或 # zadd('zz', n1=11, n2=22)zcard(name)
# 獲取name對應的有序集合元素的數量zcount(name, min, max)
# 獲取name對應的有序集合中分數 在 [min,max] 之間的個數zincrby(name, value, amount)
# 自增name對應的有序集合的 name 對應的分數r.zrange( name, start, end, desc=False, withscores=False, score_cast_func=float)
# 按照索引范圍獲取name對應的有序集合的元素 # 參數： # name，redis的name # start，有序集合索引起始位置（非分數） # end，有序集合索引結束位置（非分數） # desc，排序規則，默認按照分數從小到大排序 # withscores，是否獲取元素的分數，默認只獲取元素的值 # score_cast_func，對分數進行數據轉換的函數 # 更多： # 從大到小排序 # zrevrange(name, start, end, withscores=False, score_cast_func=float) # 按照分數范圍獲取name對應的有序集合的元素 # zrangebyscore(name, min, max, start=None, num=None, withscores=False, score_cast_func=float) # 從大到小排序 # zrevrangebyscore(name, max, min, start=None, num=None, withscores=False, score_cast_func=float)zrank(name, value)
# 獲取某個值在 name對應的有序集合中的排行（從 0 開始） # 更多： # zrevrank(name, value)，從大到小排序zrangebylex(name, min, max, start=None, num=None)
# 當有序集合的所有成員都具有相同的分值時，有序集合的元素會根據成員的 值 （lexicographical ordering）來進行排序，而這個命令則可以返回給定的有序集合鍵 key 中， 元素的值介於 min 和 max 之間的成員 # 對集合中的每個成員進行逐個字節的對比（byte-by-byte compare）， 並按照從低到高的順序， 返回排序后的集合成員。 如果兩個字符串有一部分內容是相同的話， 那么命令會認為較長的字符串比較短的字符串要大 # 參數： # name，redis的name # min，左區間（值）。 + 表示正無限； - 表示負無限； ( 表示開區間； [ 則表示閉區間 # min，右區間（值） # start，對結果進行分片處理，索引位置 # num，對結果進行分片處理，索引后面的num個元素 # 如： # ZADD myzset 0 aa 0 ba 0 ca 0 da 0 ea 0 fa 0 ga # r.zrangebylex('myzset', "-", "[ca") 結果為：['aa', 'ba', 'ca'] # 更多： # 從大到小排序 # zrevrangebylex(name, max, min, start=None, num=None)zrem(name, values)
# 刪除name對應的有序集合中值是values的成員 # 如：zrem('zz', ['s1', 's2'])zremrangebyrank(name, min, max)
# 根據排行范圍刪除zremrangebyscore(name, min, max)
# 根據分數范圍刪除zremrangebylex(name, min, max)
# 根據值返回刪除zscore(name, value)
# 獲取name對應有序集合中 value 對應的分數zinterstore(dest, keys, aggregate=None)
# 獲取兩個有序集合的交集，如果遇到相同值不同分數，則按照aggregate進行操作 # aggregate的值為: SUM MIN MAXzunionstore(dest, keys, aggregate=None)
# 獲取兩個有序集合的並集，如果遇到相同值不同分數，則按照aggregate進行操作 # aggregate的值為: SUM MIN MAXzscan(name, cursor=0, match=None, count=None, score_cast_func=float)
zscan_iter(name, match=None, count=None,score_cast_func=float)# 同字符串相似，相較於字符串新增score_cast_func，用來對分數進行操作
⑥ 其它 常用操作
delete(*names)# 根據刪除redis中的任意數據類型exists(name)
# 檢測redis的name是否存在keys(pattern='*')
# 根據模型獲取redis的name # 更多： # KEYS * 匹配數據庫中所有 key 。 # KEYS h?llo 匹配 hello ， hallo 和 hxllo 等。 # KEYS h*llo 匹配 hllo 和 heeeeello 等。 # KEYS h[ae]llo 匹配 hello 和 hallo ，但不匹配 hilloexpire(name ,time)
# 為某個redis的某個name設置超時時間rename(src, dst)
# 對redis的name重命名為move(name, db))
# 將redis的某個值移動到指定的db下randomkey()
# 隨機獲取一個redis的name（不刪除）type(name)
# 獲取name對應值的類型scan(cursor=0, match=None, count=None)
scan_iter(match=None, count=None)# 同字符串操作，用於增量迭代獲取key
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 如下：
#!/usr/bin/env python # -*- coding:utf-8 -*- import redis class RedisHelper: def __init__(self): self.__conn = redis.Redis(host='10.211.55.4') self.chan_sub = 'fm104.5' self.chan_pub = 'fm104.5' def public(self, msg): self.__conn.publish(self.chan_pub, msg) return True def subscribe(self): pub = self.__conn.pubsub() pub.subscribe(self.chan_sub) pub.parse_response() return pub
訂閱者：
#!/usr/bin/env python # -*- coding:utf-8 -*- from monitor.RedisHelper import RedisHelper obj = RedisHelper() redis_sub = obj.subscribe() while True: msg= redis_sub.parse_response() print msg
發布者：
#!/usr/bin/env python # -*- coding:utf-8 -*- from monitor.RedisHelper import RedisHelper obj = RedisHelper() obj.public('hello')
更多參見：https://github.com/andymccurdy/redis-py/
http://doc.redisfans.com/
三、RabbitMQ
1、簡介、安裝、使用
RabbitMQ 是一個在 AMQP 基礎上完成的，可復用的企業消息系統。他遵循 Mozilla Public License 開源協議。
MQ 全稱為 Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方式。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。消息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用於諸如遠程過程調用的技術。排隊指的是應用程序通過 隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。
流程上生產者把消息放到隊列中去, 然后消費者從隊列中取出消息。
- Producing , 生產者, 產生消息的角色.
- Exchange , 交換器, 在得到生產者產生的消息后, 把消息放入隊列的角色.
- Queue , 隊列, 消息暫時保存的地方.
- Consuming , 消費者, 把消息從隊列中取出的角色.
- 消息 Message
RabbitMQ安裝
# 安裝配置epel源 $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm # 安裝erlang $ yum -y install erlang # 安裝RabbitMQ $ yum -y install rabbitmq-server
# 啟動 service rabbitmq-server start/stop # 默認監聽端口5672 (帶上 SSL 默認 5671)
python 安裝 API
pip install pika or easy_install pika or 源碼 https://pypi.python.org/pypi/pika
2、使用API操作RabbitMQ
基於隊列 Queue 實現生產者消費者模型：
#!/usr/bin/env python # -*- coding:utf-8 -*- import Queue import threading message = Queue.Queue(10) def producer(i): while True: message.put(i) def consumer(i): while True: msg = message.get() for i in range(12): t = threading.Thread(target=producer, args=(i,)) t.start() for i in range(10): t = threading.Thread(target=consumer, args=(i,)) t.start()
RabbitMQ 實現：
#!/usr/bin/env python import pika # ######################### 生產者 ######################### connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.queue_declare(queue='hello') channel.basic_publish(exchange='', routing_key='hello', body='Hello World!') print(" [x] Sent 'Hello World!'") connection.close() #!/usr/bin/env python import pika # ########################## 消費者 ########################## connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print(" [x] Received %r" % body) channel.basic_consume(callback, queue='hello', no_ack=True) print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming()
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中堆積的消息會越來越多；消費者重啟后會重復消費這些消息並重復執行業務邏輯…
import pika connection = pika.BlockingConnection(pika.ConnectionParameters( host='10.211.55.4')) channel = connection.channel() channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print(" [x] Received %r" % body) import time time.sleep(10) print 'ok' ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_consume(callback, queue='hello', no_ack=False) print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming()
2、durable 消息不丟失
如果我們希望即使在RabbitMQ服務重啟的情況下，也不會丟失消息，我們可以將Queue與Message都設置為可持久化的（durable），這樣可以保證絕大部分情況下我們的RabbitMQ消息不會丟失。但依然解決不了小概率丟失事件的發生（比如RabbitMQ服務器已經接收到生產者的消息，但還沒來得及持久化該消息時RabbitMQ服務器就斷電了），如果我們需要對這種小概率事件也要管理起來，那么我們要用到事務。由於這里僅為RabbitMQ的簡單介紹，所以這里將不講解RabbitMQ相關的事務。
需要改兩處地方
#!/usr/bin/env python import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4')) channel = connection.channel() # make message persistent channel.queue_declare(queue='hello', durable=True) channel.basic_publish(exchange='', routing_key='hello', body='Hello World!', properties=pika.BasicProperties( delivery_mode=2, # make message persistent )) print(" [x] Sent 'Hello World!'") connection.close()
#!/usr/bin/env python # -*- coding:utf-8 -*- import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4')) channel = connection.channel() # make message persistent channel.queue_declare(queue='hello', durable=True) def callback(ch, method, properties, body): print(" [x] Received %r" % body) import time time.sleep(10) print 'ok' ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_consume(callback, queue='hello', no_ack=False) print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming()
3、消息獲取順序
默認情況下，消費者拿消息隊列里的數據是按平均分配，例如：消費者1 拿隊列中 奇數 序列的任務，消費者2 拿隊列中 偶數 序列的任務。
channel.basic_qos(prefetch_count=1) 表示誰來誰取，不再按照奇偶數排列，這個性能較高的機器拿的任務就多
#!/usr/bin/env python # -*- coding:utf-8 -*- import pika connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4')) channel = connection.channel() # make message persistent channel.queue_declare(queue='hello') def callback(ch, method, properties, body): print(" [x] Received %r" % body) import time time.sleep(10) print 'ok' ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(callback, queue='hello', no_ack=False) print(' [*] Waiting for messages. To exit press CTRL+C') channel.start_consuming()
4、發布訂閱
發布訂閱和簡單的消息隊列區別在於，發布訂閱者會將消息發送給所有的訂閱者，而消息隊列中的數據被消費一次便消失。所以，RabbitMQ 實現發布訂閱時，會為每一個訂閱者創建一個隊列，而發布者發布消息的時候，會將消息放置在所有相關的隊列中。
exchange type = fanout
#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', type='fanout') message = ' '.join(sys.argv[1:]) or "info: Hello World!" channel.basic_publish(exchange='logs', routing_key='', body=message) print(" [x] Sent %r" % message) connection.close()
#!/usr/bin/env python import pika connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='logs', type='fanout') result = channel.queue_declare(exclusive=True) queue_name = result.method.queue channel.queue_bind(exchange='logs', queue=queue_name) print(' [*] Waiting for logs. To exit press CTRL+C') def callback(ch, method, properties, body): print(" [x] %r" % body) channel.basic_consume(callback, queue=queue_name, no_ack=True) channel.start_consuming()
5、關鍵字發送
第4步實例中，發送消息必須明確指定某個隊列並向其中發送消息，當然，RabbitMQ 還支持根據關鍵字發送（隊列綁定關鍵字），發送者將消息發送到 exchange，exchange 根據關鍵字 判定應該將數據發送至指定隊列。
exchange type = direct
#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', type='direct') result = channel.queue_declare(exclusive=True) queue_name = result.method.queue severities = sys.argv[1:] if not severities: sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0]) sys.exit(1) for severity in severities: channel.queue_bind(exchange='direct_logs', queue=queue_name, routing_key=severity) print(' [*] Waiting for logs. To exit press CTRL+C') def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body)) channel.basic_consume(callback, queue=queue_name, no_ack=True) channel.start_consuming()
#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='direct_logs', type='direct') severity = sys.argv[1] if len(sys.argv) > 1 else 'info' message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic_publish(exchange='direct_logs', routing_key=severity, body=message) print(" [x] Sent %r:%r" % (severity, message)) connection.close()
6、模糊匹配
exchange type = topic
在 topic 類型下，可以讓隊列綁定幾個模糊的關鍵字，之后發送者將數據發送到 exchange，exchange 將傳入”路由值“和 ”關鍵字“進行匹配，匹配成功，則將數據發送到指定隊列。
匹配基本規則及示例：
- # 表示可以匹配 0 個 或 多個 單詞
- * 表示只能匹配 一個 單詞
發送者路由值 隊列中 www.suoning.python www.* -- 不匹配 www.suoning.python www.# -- 匹配
#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', type='topic') result = channel.queue_declare(exclusive=True) queue_name = result.method.queue binding_keys = sys.argv[1:] if not binding_keys: sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0]) sys.exit(1) for binding_key in binding_keys: channel.queue_bind(exchange='topic_logs', queue=queue_name, routing_key=binding_key) print(' [*] Waiting for logs. To exit press CTRL+C') def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body)) channel.basic_consume(callback, queue=queue_name, no_ack=True) channel.start_consuming()
#!/usr/bin/env python import pika import sys connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel() channel.exchange_declare(exchange='topic_logs', type='topic') routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic_publish(exchange='topic_logs', routing_key=routing_key, body=message) print(" [x] Sent %r:%r" % (routing_key, message)) connection.close()