緩存、隊列（Memcached、redis、RabbitMQ）

時間  2019-11-20
標籤 緩存 隊列 memcached redis rabbitmq 欄目 Memcached 简体版
原文   原文鏈接
  • Memcached 
    • 　　簡介、安裝、使用
    • 　　Python 操做 Memcached
    • 　　天生支持集羣
  • redis
    • 　　簡介、安裝、使用、實例
    • 　　Python 操做 Redis
    • 　　String、Hash、List、Set、Sort Set 操做
    • 　　管道
    • 　　發佈訂閱
  • RabbitMQ
    • 　　簡介、安裝、使用
    • 　　使用 API 操做 RabbitMQ
    • 　　消息不丟失
    • 　　發佈訂閱
    • 　　關鍵字發送
    • 　　模糊匹配

 

1、Memcached

一、簡介、安裝、使用

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

 

Memcached 內存管理機制：html

       Menceched 經過預分配指定的內存空間來存取數據，全部的數據都保存在 memcached 內置的內存中。python

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

　　當存入的數據佔滿內存空間時，Memcached 使用 LRU 算法自動刪除不是用的緩存數據，即重用過時數據的內存空間。Memcached 是爲緩存系統設計的，所以沒有考慮數據的容災問題，和機器的內存同樣，重啓機器將會丟失，若是但願服務重啓數據依然能保留，那麼就須要 sina 網開發的 Memcachedb 持久性內存緩衝系統，固然還有常見的 NOSQL 服務如 redis。git

默認監聽端口：11211github

 

Memcached 安裝web

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 服務安裝

# 一、安裝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 ..

# 二、安裝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 -->服務端

# 三、啓動及關閉服務
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 快速部署文檔 
# 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 PHP 客戶端

Memcached 啓動redis

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 管理sql

#一、telnet ip port 方式管理
telnet 127.0.0.1 11211

#二、命令直接操做，nc這樣的命令
[root@localhost application]# printf "stats slabs\r\n"|nc 127.0.0.1 11211    
STAT active_slabs 0
STAT total_malloced 0
END

#三、管理 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。
Memcached memadmin php工具界面化管理安裝部署文檔

 

二、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"），那麼它的執行流程以下：

  1. 將 k1 轉換成一個數字
  2. 將數字和主機列表的長度求餘數，獲得一個值 N（N 的範圍： 0 <= N < 列表長度 ）
  3. 在主機列表中根據 第2步獲得的值爲索引獲取主機，例如：host_list[N]
  4. 鏈接 將第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（獲取了緩衝的指定值），如此一來有可能出現非正常的數據，則不容許修改，並報錯。

 

 

2、redis

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

　　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

                http://www.redis.cn/

 

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
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
redis 安裝目錄及各文件做用 
# 一、 配置環境變量
# 編輯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


# 二、 拷貝配置文件
[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/


# 三、 啓動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)


# 四、 關閉redis
redis-cli shutdown
lsof -i :6379    -->檢查端口


#五、 啓動常見報錯
報錯：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
配置並啓動 redis 服務 
[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)
客戶端鏈接命令及命令測試 
# 一、下載安裝
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]#


# 二、修改php.ini設置，重啓php
在php.ini追加一條記錄
echo "extension = redis.so" >> /application/php/lib/php.ini

#重啓 php-fpm
killall php-fpm
/application/php/sbin/php-fpm

#網頁測試
......
redis 的 php 客戶端拓展安裝 
# 一、修改從庫redis.conf配置文件
#配置從庫redis.conf配置文件(先裝redis)
#添加一行，主庫IP地址及端口
vim /application/redis/conf/redis.conf
# slaveof <masterip> <masterport>
slaveof 192.168.200.95 6379


# 二、重啓從庫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  -->成功


# 三、測試主從同步
# 主庫：寫數據
[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 主從同步 
至於 redis 的負載均衡，方案有不少：
LVS、keepalived、Twemproxy
小編有時間再補上吧...
reidis 負載均衡 
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文件恢復數據
redis 持久化

 

 

二、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,自增數（必須是整數）
 
# 注：同incrb

incrbyfloat(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獲取value

hmget(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是否存在當前傳入的key

hdel(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=amount

hscan(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，要添加數據的列表的name

blpop(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對應的列表的全部元素，那麼就須要：
    # 一、獲取name對應的全部列表
    # 二、循環列表
# 可是，若是列表很是大，那麼就有可能在第一步時就將程序的內容撐爆，全部有必要自定義一個增量迭代的功能：
 
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  MAX

zunionstore(dest, keys, aggregate=None)

# 獲取兩個有序集合的並集，若是遇到相同值不一樣分數，則按照aggregate進行操做
# aggregate的值爲:  SUM  MIN  MAX

zscan(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 ，但不匹配 hillo

expire(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
RedisHelper

訂閱者：

#!/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/

 

 

3、RabbitMQ

一、簡介、安裝、使用

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

 

二、使用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()
View Code

 

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()

 

 

一、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()
消費者

二、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()
消費者

三、消息獲取順序

默認狀況下，消費者拿消息隊列裏的數據是按平均分配，例如：消費者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()
消費者

四、發佈訂閱

發佈訂閱和簡單的消息隊列區別在於，發佈訂閱者會將消息發送給全部的訂閱者，而消息隊列中的數據被消費一次便消失。因此，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()
訂閱者

五、關鍵字發送

第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()
生產者

六、模糊匹配

 

 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()
生產者

 

 
 來自：http://www.cnblogs.com/suoning/p/5807247.html
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程