day9-Memcached & Redis使用

Memcached

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

Memcached安裝和基本使用

Memcached安裝：

1 2 3 4 5 6 7 8

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 2 3 4 5 6 7 8 9 10

memcached  - d  - m  10     - u root  - l  10.211 . 55.4  - 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命令

1 2 3

存儲命令:  set / add / replace / append / prepend / cas 獲取命令: get / gets 其餘命令: delete / stats..

Python操做Memcached

安裝API

1 2	python操做Memcached使用Python - memcached模塊 下載安裝：https: / / pypi.python.org / pypi / python - memcached

一、第一次操做

1 2 3 4 5 6

import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True ) mc. set ( "foo" ,  "bar" ) ret  =  mc.get( 'foo' ) print  ret

Ps：debug = True 表示運行出現錯誤時，現實錯誤信息，上線後移除該參數。

二、天生支持集羣

python-memcached模塊原生支持集羣操做，其原理是在內存維護一個主機列表，且集羣中主機的權重值和主機在列表中重複出現的次數成正比

1 2 3 4 5 6 7

主機    權重      1.1 . 1.1    1      1.1 . 1.2    2      1.1 . 1.3    1   那麼在內存中主機列表爲：      host_list  =  [ "1.1.1.1" ,  "1.1.1.2" ,  "1.1.1.2" ,  "1.1.1.3" , ]

若是用戶根據若是要在內存中建立一個鍵值對（如：k1 = "v1"），那麼要執行一下步驟：

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

代碼實現以下：

1 2 3

mc  =  memcache.Client([( '1.1.1.1:12000' ,  1 ), ( '1.1.1.2:12000' ,  2 ), ( '1.1.1.3:12000' ,  1 )], debug = True )   mc. set ( 'k1' ,  'v1' )

三、add
添加一條鍵值對，若是已經存在的 key，重複執行add操做異常

1 2 3 4 5 6 7

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True ) mc.add( 'k1' ,  'v1' ) # mc.add('k1', 'v2') # 報錯，對已經存在的key重複添加，失敗！！！

四、replace
replace 修改某個key的值，若是key不存在，則異常

1 2 3 4 5 6 7

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True ) # 若是memcache中存在kkkk，則替換成功，不然一場 mc.replace( 'kkkk' , '999' )

五、set 和 set_multi

set            設置一個鍵值對，若是key不存在，則建立，若是key存在，則修改
set_multi   設置多個鍵值對，若是key不存在，則建立，若是key存在，則修改

1 2 3 4 5 6 7 8 9

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True )   mc. set ( 'key0' ,  'wupeiqi' )   mc.set_multi({ 'key1' :  'val1' ,  'key2' :  'val2' })

六、delete 和 delete_multi

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

1 2 3 4 5 6 7 8

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True )   mc.delete( 'key0' ) mc.delete_multi([ 'key1' ,  'key2' ])

七、get 和 get_multi

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

1 2 3 4 5 6 7 8

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True )   val  =  mc.get( 'key0' ) item_dict  =  mc.get_multi([ "key1" ,  "key2" ,  "key3" ])

八、append 和 prepend

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

1 2 3 4 5 6 7 8 9 10 11 12

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True ) # k1 = "v1"   mc.append( 'k1' ,  'after' ) # k1 = "v1after"   mc.prepend( 'k1' ,  'before' ) # k1 = "beforev1after"

九、decr 和 incr　　

incr  自增，將Memcached中的某一個值增長 N （ N默認爲1 ）
decr 自減，將Memcached中的某一個值減小 N （ N默認爲1 ）

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache   mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True ) mc. set ( 'k1' ,  '777' )   mc.incr( 'k1' ) # k1 = 778   mc.incr( 'k1' ,  10 ) # k1 = 788   mc.decr( 'k1' ) # k1 = 787   mc.decr( 'k1' ,  10 ) # k1 = 777

十、gets 和 cas

如商城商品剩餘個數，假設改值保存在memcache中，product_count = 900
A用戶刷新頁面從memcache中讀取到product_count = 900
B用戶刷新頁面從memcache中讀取到product_count = 900

若是A、B用戶均購買商品

A用戶修改商品剩餘個數 product_count＝899
B用戶修改商品剩餘個數 product_count＝899

如此一來緩存內的數據便不在正確，兩個用戶購買商品後，商品剩餘仍是 899
若是使用python的set和get來操做以上過程，那麼程序就會如上述所示狀況！

若是想要避免此狀況的發生，只要使用 gets 和 cas 便可，如：

1 2 3 4 5 6 7 8 9

#!/usr/bin/env python # -*- coding:utf-8 -*- import  memcache mc  =  memcache.Client([ '10.211.55.4:12000' ], debug = True , cache_cas = True )   v  =  mc.gets( 'product_count' ) # ... # 若是有人在gets以後和cas以前修改了product_count，那麼，下面的設置將會執行失敗，剖出異常，從而避免非正常數據的產生 mc.cas( 'product_count' ,  "899" )

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

Memcached 真的過期了嗎？

Redis

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

 

  1 1. 使用Redis有哪些好處？
  2 
  3 (1) 速度快，由於數據存在內存中，相似於HashMap，HashMap的優點就是查找和操做的時間複雜度都是O(1)
  4 
  5 (2) 支持豐富數據類型，支持string，list，set，sorted set，hash
  6 
  7 (3) 支持事務，操做都是原子性，所謂的原子性就是對數據的更改要麼所有執行，要麼所有不執行
  8 
  9 (4) 豐富的特性：可用於緩存，消息，按key設置過時時間，過時後將會自動刪除
 10 
 11 
 12 2. redis相比memcached有哪些優點？
 13 
 14 (1) memcached全部的值均是簡單的字符串，redis做爲其替代者，支持更爲豐富的數據類型
 15 
 16 (2) redis的速度比memcached快不少
 17 
 18 (3) redis能夠持久化其數據
 19 
 20 
 21 3. redis常見性能問題和解決方案：
 22 
 23 (1) Master最好不要作任何持久化工做，如RDB內存快照和AOF日誌文件
 24 
 25 (2) 若是數據比較重要，某個Slave開啓AOF備份數據，策略設置爲每秒同步一次
 26 
 27 (3) 爲了主從複製的速度和鏈接的穩定性，Master和Slave最好在同一個局域網內
 28 
 29 (4) 儘可能避免在壓力很大的主庫上增長從庫
 30 
 31 (5) 主從複製不要用圖狀結構，用單向鏈表結構更爲穩定，即：Master <- Slave1 <- Slave2 <- Slave3...
 32 
 33 這樣的結構方便解決單點故障問題，實現Slave對Master的替換。若是Master掛了，能夠馬上啓用Slave1作Master，其餘不變。
 34 
 35 
 36 
 37  
 38 
 39 4. MySQL裏有2000w數據，redis中只存20w的數據，如何保證redis中的數據都是熱點數據
 40 
 41  相關知識：redis 內存數據集大小上升到必定大小的時候，就會施行數據淘汰策略。redis 提供 6種數據淘汰策略：
 42 
 43 voltile-lru：從已設置過時時間的數據集（server.db[i].expires）中挑選最近最少使用的數據淘汰
 44 
 45 volatile-ttl：從已設置過時時間的數據集（server.db[i].expires）中挑選將要過時的數據淘汰
 46 
 47 volatile-random：從已設置過時時間的數據集（server.db[i].expires）中任意選擇數據淘汰
 48 
 49 allkeys-lru：從數據集（server.db[i].dict）中挑選最近最少使用的數據淘汰
 50 
 51 allkeys-random：從數據集（server.db[i].dict）中任意選擇數據淘汰
 52 
 53 no-enviction（驅逐）：禁止驅逐數據
 54 
 55  
 56 
 57 5. Memcache與Redis的區別都有哪些？
 58 
 59 1)、存儲方式
 60 
 61 Memecache把數據所有存在內存之中，斷電後會掛掉，數據不能超過內存大小。
 62 
 63 Redis有部份存在硬盤上，這樣能保證數據的持久性。
 64 
 65 2)、數據支持類型
 66 
 67 Memcache對數據類型支持相對簡單。
 68 
 69 Redis有複雜的數據類型。
 70 
 71 
 72 3），value大小
 73 
 74 redis最大能夠達到1GB，而memcache只有1MB
 75 
 76 
 77 
 78 6. Redis 常見的性能問題都有哪些？如何解決？
 79 
 80  
 81 
 82 1).Master寫內存快照，save命令調度rdbSave函數，會阻塞主線程的工做，當快照比較大時對性能影響是很是大的，會間斷性暫停服務，因此Master最好不要寫內存快照。
 83 
 84 
 85 2).Master AOF持久化，若是不重寫AOF文件，這個持久化方式對性能的影響是最小的，可是AOF文件會不斷增大，AOF文件過大會影響Master重啓的恢復速度。Master最好不要作任何持久化工做，包括內存快照和AOF日誌文件，特別是不要啓用內存快照作持久化,若是數據比較關鍵，某個Slave開啓AOF備份數據，策略爲每秒同步一次。
 86 
 87  
 88 3).Master調用BGREWRITEAOF重寫AOF文件，AOF在重寫的時候會佔大量的CPU和內存資源，致使服務load太高，出現短暫服務暫停現象。
 89 
 90 4). Redis主從複製的性能問題，爲了主從複製的速度和鏈接的穩定性，Slave和Master最好在同一個局域網內
 91 
 92 
 93 
 94 
 95 7, redis 最適合的場景
 96 
 97 
 98 Redis最適合全部數據in-momory的場景，雖然Redis也提供持久化功能，但實際更多的是一個disk-backed的功能，跟傳統意義上的持久化有比較大的差異，那麼可能你們就會有疑問，彷佛Redis更像一個增強版的Memcached，那麼什麼時候使用Memcached,什麼時候使用Redis呢?
 99 
100        若是簡單地比較Redis與Memcached的區別，大多數都會獲得如下觀點：
101 、Redis不只僅支持簡單的k/v類型的數據，同時還提供list，set，zset，hash等數據結構的存儲。
102 、Redis支持數據的備份，即master-slave模式的數據備份。
103 、Redis支持數據的持久化，能夠將內存中的數據保持在磁盤中，重啓的時候能夠再次加載進行使用。
104 
105 （1）、會話緩存（Session Cache）
106 
107 最經常使用的一種使用Redis的情景是會話緩存（session cache）。用Redis緩存會話比其餘存儲（如Memcached）的優點在於：Redis提供持久化。當維護一個不是嚴格要求一致性的緩存時，若是用戶的購物車信息所有丟失，大部分人都會不高興的，如今，他們還會這樣嗎？
108 
109 幸運的是，隨着 Redis 這些年的改進，很容易找到怎麼恰當的使用Redis來緩存會話的文檔。甚至廣爲人知的商業平臺Magento也提供Redis的插件。
110 
111 （2）、全頁緩存（FPC）
112 
113 除基本的會話token以外，Redis還提供很簡便的FPC平臺。回到一致性問題，即便重啓了Redis實例，由於有磁盤的持久化，用戶也不會看到頁面加載速度的降低，這是一個極大改進，相似PHP本地FPC。
114 
115 再次以Magento爲例，Magento提供一個插件來使用Redis做爲全頁緩存後端。
116 
117 此外，對WordPress的用戶來講，Pantheon有一個很是好的插件  wp-redis，這個插件能幫助你以最快速度加載你曾瀏覽過的頁面。
118 
119 （3）、隊列
120 
121 Reids在內存存儲引擎領域的一大優勢是提供 list 和 set 操做，這使得Redis能做爲一個很好的消息隊列平臺來使用。Redis做爲隊列使用的操做，就相似於本地程序語言（如Python）對 list 的 push/pop 操做。
122 
123 若是你快速的在Google中搜索「Redis queues」，你立刻就能找到大量的開源項目，這些項目的目的就是利用Redis建立很是好的後端工具，以知足各類隊列需求。例如，Celery有一個後臺就是使用Redis做爲broker，你能夠從這裏去查看。
124 
125 （4），排行榜/計數器
126 
127 Redis在內存中對數字進行遞增或遞減的操做實現的很是好。集合（Set）和有序集合（Sorted Set）也使得咱們在執行這些操做的時候變的很是簡單，Redis只是正好提供了這兩種數據結構。因此，咱們要從排序集合中獲取到排名最靠前的10個用戶–咱們稱之爲「user_scores」，咱們只須要像下面同樣執行便可：
128 
129 固然，這是假定你是根據你用戶的分數作遞增的排序。若是你想返回用戶及用戶的分數，你須要這樣執行：
130 
131 ZRANGE user_scores 0 10 WITHSCORES
132 
133 Agora Games就是一個很好的例子，用Ruby實現的，它的排行榜就是使用Redis來存儲數據的，你能夠在這裏看到。
134 
135 （5）、發佈/訂閱
136 
137 最後（但確定不是最不重要的）是Redis的發佈/訂閱功能。發佈/訂閱的使用場景確實很是多。我已看見人們在社交網絡鏈接中使用，還可做爲基於發佈/訂閱的腳本觸發器，甚至用Redis的發佈/訂閱功能來創建聊天系統！（不，這是真的，你能夠去核實）。
138 
139 Redis提供的全部特性中，我感受這個是喜歡的人最少的一個，雖然它爲用戶提供若是此多功能。

View Code

 

1、Redis安裝和基本使用

1 2 3 4

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

啓動服務端

1	src / redis - server

啓動客戶端

1 2 3 4 5

src / redis - cli redis>  set  foo bar OK redis> get foo "bar"

2、Python操做Redis

1 2 3 4 5 6 7

sudo pip install redis or sudo easy_install redis or 源碼安裝   詳見：https: / / github.com / WoLpH / redis - py

API使用

redis-py 的API的使用能夠分類爲：

• 鏈接方式
• 鏈接池
• 操做
  • String 操做
  • Hash 操做
  • List 操做
  • Set 操做
  • Sort Set 操做
• 管道
• 發佈訂閱

 

一、操做模式

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

1 2 3 4 5 6 7 8

#!/usr/bin/env python # -*- coding:utf-8 -*-   import  redis   r  =  redis.Redis(host = '10.211.55.4' , port = 6379 ) r. set ( 'foo' ,  'Bar' ) print  r.get( 'foo' )

二、鏈接池

redis-py使用connection pool來管理對一個redis server的全部鏈接，避免每次創建、釋放鏈接的開銷。默認，每一個Redis實例都會維護一個本身的鏈接池。能夠直接創建一個鏈接池，而後做爲參數Redis，這樣就能夠實現多個Redis實例共享一個鏈接池。

1 2 3 4 5 6 7 8 9 10

#!/usr/bin/env python # -*- coding:utf-8 -*-   import  redis   pool  =  redis.ConnectionPool(host = '10.211.55.4' , port = 6379 )   r  =  redis.Redis(connection_pool = pool) r. set ( 'foo' ,  'Bar' ) print  r.get( 'foo' )

三、操做

String操做，redis中的String在在內存中按照一個name對應一個value來存儲。如圖：

set(name, value, ex=None, px=None, nx=False, xx=False)

1 2 3 4 5 6

在Redis中設置值，默認，不存在則建立，存在則修改 參數：       ex，過時時間（秒）       px，過時時間（毫秒）       nx，若是設置爲True，則只有name不存在時，當前set操做才執行       xx，若是設置爲True，則只有name存在時，崗前set操做才執行

setnx(name, value)

1	設置值，只有name不存在時，執行設置操做（添加）

setex(name, value, time)

1 2 3

# 設置值 # 參數：      # time，過時時間（數字秒 或 timedelta對象）

psetex(name, time_ms, value)

1 2 3

# 設置值 # 參數：      # time_ms，過時時間（數字毫秒 或 timedelta對象）

mset(*args, **kwargs)

1 2 3 4 5

批量設置值 如：      mset(k1= 'v1' , k2= 'v2' )      或      mget({ 'k1' :  'v1' ,  'k2' :  'v2' })

get(name)

1

獲取值

mget(keys, *args)

1 2 3 4 5

批量獲取 如：      mget( 'ylr' ,  'wupeiqi' )      或      r.mget([ 'ylr' ,  'wupeiqi' ])

getset(name, value)

1	設置新值並獲取原來的值

getrange(key, start, end)

1 2 3 4 5 6

# 獲取子序列（根據字節獲取，非字符） # 參數：      # name，Redis 的 name      # start，起始位置（字節）      # end，結束位置（字節） # 如： "武沛齊" ，0-3表示 "武"

setrange(name, offset, value)

1 2 3 4

# 修改字符串內容，從指定字符串索引開始向後替換（新值太長時，則向後添加） # 參數：      # offset，字符串的索引，字節（一個漢字三個字節）      # value，要設置的值

setbit(name, offset, value)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27

# 對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"   # 擴展，轉換二進制表示：        # source = "武沛齊"      source  =  "foo"        for  i  in  source:          num  =  ord (i)          print  bin (num).replace( 'b' ,'')        特別的，若是source是漢字  "武沛齊" 怎麼辦？      答：對於utf - 8 ，每個漢字佔  3  個字節，那麼  "武沛齊"  則有  9 個字節         對於漢字， for 循環時候會按照 字節 迭代，那麼在迭代時，將每個字節轉換 十進制數，而後再將十進制數轉換成二進制          11100110  10101101  10100110  11100110  10110010  10011011  11101001  10111101  10010000          - - - - - - - - - - - - - - - - - - - - - - - - - -  - - - - - - - - - - - - - - - - - - - - - - - - - - - - -  - - - - - - - - - - - - - - - - - - - - - - - - - - - - -                      武                         沛                           齊

getbit(name, offset)

1	# 獲取name對應的值的二進制表示中的某位的值 （0或1）

bitcount(key, start=None, end=None)

1 2 3 4 5

# 獲取name對應的值的二進制表示中 1 的個數 # 參數：      # key，Redis的name      # start，位起始位置      # end，位結束位置

bitop(operation, dest, *keys)

1 2 3 4 5 6 7 8 9 10

# 獲取多個值，並將值作位運算，將最後的結果保存至新的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)

1	# 返回name對應值的字節長度（一個漢字3個字節）

incr(self, name, amount=1)

1 2 3 4 5 6 7

# 自增 name對應的值，當name不存在時，則建立name＝amount，不然，則自增。   # 參數：      # name,Redis的name      # amount,自增數（必須是整數）   # 注：同incrby

incrbyfloat(self, name, amount=1.0)

1 2 3 4 5

# 自增 name對應的值，當name不存在時，則建立name＝amount，不然，則自增。   # 參數：      # name,Redis的name      # amount,自增數（浮點型）

decr(self, name, amount=1)

1 2 3 4 5

# 自減 name對應的值，當name不存在時，則建立name＝amount，不然，則自減。   # 參數：      # name,Redis的name      # amount,自減數（整數）

append(key, value)

1 2 3 4 5

# 在redis name對應的值後面追加內容   # 參數：      key, redis的name      value, 要追加的字符串

　　

Hash操做，redis中Hash在內存中的存儲格式以下圖：

hset(name, key, value)

1 2 3 4 5 6 7 8 9

# name對應的hash中設置一個鍵值對（不存在，則建立；不然，修改）   # 參數：      # name，redis的name      # key，name對應的hash中的key      # value，name對應的hash中的value   # 注：      # hsetnx(name, key, value),當name對應的hash中不存在當前key時則建立（至關於添加）

hmset(name, mapping)

1 2 3 4 5 6 7 8

# 在name對應的hash中批量設置鍵值對   # 參數：      # name，redis的name      # mapping，字典，如：{'k1':'v1', 'k2': 'v2'}   # 如：      # r.hmset('xx', {'k1':'v1', 'k2': 'v2'})

hget(name,key)

1	# 在name對應的hash中獲取根據key獲取value

hmget(name, keys, *args)

1 2 3 4 5 6 7 8 9 10 11

# 在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)

1	獲取name對應 hash 的全部鍵值

hlen(name)

1	# 獲取name對應的hash中鍵值對的個數

hkeys(name)

1	# 獲取name對應的hash中全部的key的值

hvals(name)

1	# 獲取name對應的hash中全部的value的值

hexists(name, key)

1	# 檢查name對應的hash是否存在當前傳入的key

hdel(name,*keys)

1	# 將name對應的hash中指定key的鍵值對刪除

hincrby(name, key, amount=1)

1 2 3 4 5

# 自增name對應的hash中的指定key的值，不存在則建立key=amount # 參數：      # name，redis中的name      # key， hash對應的key      # amount，自增數（整數）

hincrbyfloat(name, key, amount=1.0)

1 2 3 4 5 6 7 8

# 自增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)

1 2 3 4 5 6 7 8 9 10 11 12 13

# 增量式迭代獲取，對於數據大的數據很是有用，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)

1 2 3 4 5 6 7 8 9

# 利用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)

1 2 3 4 5 6 7 8

# 在name對應的list中添加元素，每一個新的元素都添加到列表的最左邊   # 如：      # r.lpush('oo', 11,22,33)      # 保存順序爲: 33,22,11   # 擴展：      # rpush(name, values) 表示從右向左操做

lpushx(name,value)

1 2 3 4

# 在name對應的list中添加元素，只有name已經存在時，值添加到列表的最左邊   # 更多：      # rpushx(name, value) 表示從右向左操做

llen(name)

1	# name對應的list元素的個數

linsert(name, where, refvalue, value))

1 2 3 4 5 6 7

# 在name對應的列表的某一個值前或後插入一個新值   # 參數：      # name，redis的name      # where，BEFORE或AFTER      # refvalue，標杆值，即：在它先後插入數據      # value，要插入的數據

r.lset(name, index, value)

1 2 3 4 5 6

# 對name對應的list中的某一個索引位置從新賦值   # 參數：      # name，redis的name      # index，list的索引位置      # value，要設置的值

r.lrem(name, value, num)

1 2 3 4 5 6 7 8

# 在name對應的list中刪除指定的值   # 參數：      # name，redis的name      # value，要刪除的值      # num，  num=0，刪除列表中全部的指定值；             # num=2,從前到後，刪除2個；             # num=-2,從後向前，刪除2個

lpop(name)

1 2 3 4

# 在name對應的列表的左側獲取第一個元素並在列表中移除，返回值則是第一個元素   # 更多：      # rpop(name) 表示從右向左操做

lindex(name, index)

1	在name對應的列表中根據索引獲取列表元素

lrange(name, start, end)

1 2 3 4 5

# 在name對應的列表分片獲取數據 # 參數：      # name，redis的name      # start，索引的起始位置      # end，索引結束位置

ltrim(name, start, end)

1 2 3 4 5

# 在name對應的列表中移除沒有在start-end索引之間的值 # 參數：      # name，redis的name      # start，索引的起始位置      # end，索引結束位置

rpoplpush(src, dst)

1 2 3 4

# 從一個列表取出最右邊的元素，同時將其添加至另外一個列表的最左邊 # 參數：      # src，要取數據的列表的name      # dst，要添加數據的列表的name

blpop(keys, timeout)

1 2 3 4 5 6 7 8

# 將多個列表排列，按照從左到右去pop對應列表的元素   # 參數：      # keys，redis的name的集合      # timeout，超時時間，當元素全部列表的元素獲取完以後，阻塞等待列表內有數據的時間（秒）, 0 表示永遠阻塞   # 更多：      # r.brpop(keys, timeout)，從右向左獲取數據

brpoplpush(src, dst, timeout=0)

1 2 3 4 5 6

# 從一個列表的右側移除一個元素並將其添加到另外一個列表的左側   # 參數：      # src，取出並要移除元素的列表對應的name      # dst，要插入元素的列表對應的name      # timeout，當src對應的列表中沒有數據時，阻塞等待其有數據的超時時間（秒），0 表示永遠阻塞

自定義增量迭代

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

# 因爲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)

1	# name對應的集合中添加元素

scard(name)

1	獲取name對應的集合中元素個數

sdiff(keys, *args)

1	在第一個name對應的集合中且不在其餘name對應的集合的元素集合

sdiffstore(dest, keys, *args)

1	# 獲取第一個name對應的集合中且不在其餘name對應的集合，再將其新加入到dest對應的集合中

sinter(keys, *args)

1	# 獲取多一個name對應集合的並集

sinterstore(dest, keys, *args)

1	# 獲取多一個name對應集合的並集，再講其加入到dest對應的集合中

sismember(name, value)

1	# 檢查value是不是name對應的集合的成員

smembers(name)

1	# 獲取name對應的集合的全部成員

smove(src, dst, value)

1	# 將某個成員從一個集合中移動到另一個集合

spop(name)

1	# 從集合的右側（尾部）移除一個成員，並將其返回

srandmember(name, numbers)

1	# 從name對應的集合中隨機獲取 numbers 個元素

srem(name, values)

1	# 在name對應的集合中刪除某些值

sunion(keys, *args)

1	# 獲取多一個name對應的集合的並集

sunionstore(dest,keys, *args)

1	# 獲取多一個name對應的集合的並集，並將結果保存到dest對應的集合中

sscan(name, cursor=0, match=None, count=None)
sscan_iter(name, match=None, count=None)

1	# 同字符串的操做，用於增量迭代分批獲取元素，避免內存消耗太大

 

有序集合，在集合的基礎上，爲每元素排序；元素的排序須要根據另一個值來進行比較，因此，對於有序集合，每個元素有兩個值，即：值和分數，分數專門用來作排序。

zadd(name, *args, **kwargs)

1 2 3 4 5

# 在name對應的有序集合中添加元素 # 如：       # zadd('zz', 'n1', 1, 'n2', 2)       # 或       # zadd('zz', n1=11, n2=22)

zcard(name)

1	# 獲取name對應的有序集合元素的數量

zcount(name, min, max)

1	# 獲取name對應的有序集合中分數 在 [min,max] 之間的個數

zincrby(name, value, amount)

1	# 自增name對應的有序集合的 name 對應的分數

r.zrange( name, start, end, desc=False, withscores=False, score_cast_func=float)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

# 按照索引範圍獲取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)

1 2 3 4

# 獲取某個值在 name對應的有序集合中的排行（從 0 開始）   # 更多：      # zrevrank(name, value)，從大到小排序

zrangebylex(name, min, max, start=None, num=None)

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

# 當有序集合的全部成員都具備相同的分值時，有序集合的元素會根據成員的 值 （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)

1 2 3

# 刪除name對應的有序集合中值是values的成員   # 如：zrem('zz', ['s1', 's2'])

zremrangebyrank(name, min, max)

1	# 根據排行範圍刪除

zremrangebyscore(name, min, max)

1	# 根據分數範圍刪除

zremrangebylex(name, min, max)

1

# 根據值返回刪除

zscore(name, value)

1	# 獲取name對應有序集合中 value 對應的分數

zinterstore(dest, keys, aggregate=None)

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

zunionstore(dest, keys, aggregate=None)

1 2	# 獲取兩個有序集合的並集，若是遇到相同值不一樣分數，則按照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)

1	# 同字符串類似，相較於字符串新增score_cast_func，用來對分數進行操做

　　

其餘經常使用操做

delete(*names)

1	# 根據刪除redis中的任意數據類型

exists(name)

1	# 檢測redis的name是否存在

keys(pattern='*')

1 2 3 4 5 6 7

# 根據模型獲取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)

1	# 爲某個redis的某個name設置超時時間

rename(src, dst)

1	# 對redis的name重命名爲

move(name, db))

1	# 將redis的某個值移動到指定的db下

randomkey()

1	# 隨機獲取一個redis的name（不刪除）

type(name)

1	# 獲取name對應值的類型

scan(cursor=0, match=None, count=None)
scan_iter(match=None, count=None)

1	# 同字符串操做，用於增量迭代獲取key

　

四、管道

redis-py默認在執行每次請求都會建立（鏈接池申請鏈接）和斷開（歸還鏈接池）一次鏈接操做，若是想要在一次請求中指定多個命令，則可使用pipline實現一次請求指定多個命令，而且默認狀況下一次pipline 是原子性操做。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

#!/usr/bin/env python # -*- coding:utf-8 -*-   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 )   pipe. set ( 'name' ,  'alex' ) pipe. set ( 'role' ,  'sb' )   pipe.execute()

五、發佈訂閱

發佈者：服務器

訂閱者：Dashboad和數據處理

Demo以下：

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 
 4 import redis
 5 
 6 
 7 class RedisHelper:
 8 
 9     def __init__(self):
10         self.__conn = redis.Redis(host='10.211.55.4')
11         self.chan_sub = 'fm104.5'
12         self.chan_pub = 'fm104.5'
13 
14     def public(self, msg):
15         self.__conn.publish(self.chan_pub, msg)
16         return True
17 
18     def subscribe(self):
19         pub = self.__conn.pubsub()
20         pub.subscribe(self.chan_sub)
21         pub.parse_response()
22         return pub

RedisHelper

 

訂閱者：

1 2 3 4 5 6 7 8 9 10 11

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

發佈者：

1 2 3 4 5 6 7

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

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

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

RabbitMQ安裝

1 2 3 4 5 6 7 8

安裝配置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

安裝API

1 2 3 4 5 6 7

pip install pika or easy_install pika or 源碼   https: / / pypi.python.org / pypi / pika

使用API操做RabbitMQ

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

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 import Queue
 4 import threading
 5 
 6 
 7 message = Queue.Queue(10)
 8 
 9 
10 def producer(i):
11     while True:
12         message.put(i)
13 
14 
15 def consumer(i):
16     while True:
17         msg = message.get()
18 
19 
20 for i in range(12):
21     t = threading.Thread(target=producer, args=(i,))
22     t.start()
23 
24 for i in range(10):
25     t = threading.Thread(target=consumer, args=(i,))
26     t.start()

View Code

 

對於RabbitMQ來講，生產和消費再也不針對內存裏的一個Queue對象，而是某臺服務器上的RabbitMQ Server實現的消息隊列。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

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

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

#!/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會從新將該任務添加到隊列中。

 1 import pika
 2 
 3 connection = pika.BlockingConnection(pika.ConnectionParameters(
 4         host='10.211.55.4'))
 5 channel = connection.channel()
 6 
 7 channel.queue_declare(queue='hello')
 8 
 9 def callback(ch, method, properties, body):
10     print(" [x] Received %r" % body)
11     import time
12     time.sleep(10)
13     print 'ok'
14     ch.basic_ack(delivery_tag = method.delivery_tag)
15 
16 channel.basic_consume(callback,
17                       queue='hello',
18                       no_ack=False)
19 
20 print(' [*] Waiting for messages. To exit press CTRL+C')
21 channel.start_consuming()

消費者

二、durable   消息不丟失

 1 #!/usr/bin/env python
 2 import pika
 3 
 4 connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
 5 channel = connection.channel()
 6 
 7 # make message persistent
 8 channel.queue_declare(queue='hello', durable=True)
 9 
10 channel.basic_publish(exchange='',
11                       routing_key='hello',
12                       body='Hello World!',
13                       properties=pika.BasicProperties(
14                           delivery_mode=2, # make message persistent
15                       ))
16 print(" [x] Sent 'Hello World!'")
17 connection.close()

生產者

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 import pika
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
 6 channel = connection.channel()
 7 
 8 # make message persistent
 9 channel.queue_declare(queue='hello', durable=True)
10 
11 
12 def callback(ch, method, properties, body):
13     print(" [x] Received %r" % body)
14     import time
15     time.sleep(10)
16     print 'ok'
17     ch.basic_ack(delivery_tag = method.delivery_tag)
18 
19 channel.basic_consume(callback,
20                       queue='hello',
21                       no_ack=False)
22 
23 print(' [*] Waiting for messages. To exit press CTRL+C')
24 channel.start_consuming()

消費者

 

三、消息獲取順序

默認消息隊列裏的數據是按照順序被消費者拿走，例如：消費者1 去隊列中獲取 奇數 序列的任務，消費者1去隊列中獲取 偶數 序列的任務。

channel.basic_qos(prefetch_count=1) 表示誰來誰取，再也不按照奇偶數排列

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 import pika
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.211.55.4'))
 6 channel = connection.channel()
 7 
 8 # make message persistent
 9 channel.queue_declare(queue='hello')
10 
11 
12 def callback(ch, method, properties, body):
13     print(" [x] Received %r" % body)
14     import time
15     time.sleep(10)
16     print 'ok'
17     ch.basic_ack(delivery_tag = method.delivery_tag)
18 
19 channel.basic_qos(prefetch_count=1)
20 
21 channel.basic_consume(callback,
22                       queue='hello',
23                       no_ack=False)
24 
25 print(' [*] Waiting for messages. To exit press CTRL+C')
26 channel.start_consuming()

消費者

 

四、發佈訂閱

發佈訂閱和簡單的消息隊列區別在於，發佈訂閱會將消息發送給全部的訂閱者，而消息隊列中的數據被消費一次便消失。因此，RabbitMQ實現發佈和訂閱時，會爲每個訂閱者建立一個隊列，而發佈者發佈消息時，會將消息放置在全部相關隊列中。

 exchange type = fanout

 1 #!/usr/bin/env python
 2 import pika
 3 import sys
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6         host='localhost'))
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange='logs',
10                          type='fanout')
11 
12 message = ' '.join(sys.argv[1:]) or "info: Hello World!"
13 channel.basic_publish(exchange='logs',
14                       routing_key='',
15                       body=message)
16 print(" [x] Sent %r" % message)
17 connection.close()

發佈者

 1 #!/usr/bin/env python
 2 import pika
 3 
 4 connection = pika.BlockingConnection(pika.ConnectionParameters(
 5         host='localhost'))
 6 channel = connection.channel()
 7 
 8 channel.exchange_declare(exchange='logs',
 9                          type='fanout')
10 
11 result = channel.queue_declare(exclusive=True)
12 queue_name = result.method.queue
13 
14 channel.queue_bind(exchange='logs',
15                    queue=queue_name)
16 
17 print(' [*] Waiting for logs. To exit press CTRL+C')
18 
19 def callback(ch, method, properties, body):
20     print(" [x] %r" % body)
21 
22 channel.basic_consume(callback,
23                       queue=queue_name,
24                       no_ack=True)
25 
26 channel.start_consuming()

訂閱者

 

五、關鍵字發送

 exchange type = direct

以前事例，發送消息時明確指定某個隊列並向其中發送消息，RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 斷定應該將數據發送至指定隊列。

 1 #!/usr/bin/env python
 2 import pika
 3 import sys
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6         host='localhost'))
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange='direct_logs',
10                          type='direct')
11 
12 result = channel.queue_declare(exclusive=True)
13 queue_name = result.method.queue
14 
15 severities = sys.argv[1:]
16 if not severities:
17     sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
18     sys.exit(1)
19 
20 for severity in severities:
21     channel.queue_bind(exchange='direct_logs',
22                        queue=queue_name,
23                        routing_key=severity)
24 
25 print(' [*] Waiting for logs. To exit press CTRL+C')
26 
27 def callback(ch, method, properties, body):
28     print(" [x] %r:%r" % (method.routing_key, body))
29 
30 channel.basic_consume(callback,
31                       queue=queue_name,
32                       no_ack=True)
33 
34 channel.start_consuming()

消費者

 1 #!/usr/bin/env python
 2 import pika
 3 import sys
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6         host='localhost'))
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange='direct_logs',
10                          type='direct')
11 
12 severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
13 message = ' '.join(sys.argv[2:]) or 'Hello World!'
14 channel.basic_publish(exchange='direct_logs',
15                       routing_key=severity,
16                       body=message)
17 print(" [x] Sent %r:%r" % (severity, message))
18 connection.close()

生產者

 

六、模糊匹配

 exchange type = topic

在topic類型下，可讓隊列綁定幾個模糊的關鍵字，以後發送者將數據發送到exchange，exchange將傳入」路由值「和 」關鍵字「進行匹配，匹配成功，則將數據發送到指定隊列。

• # 表示能夠匹配 0 個 或 多個 單詞
• *  表示只能匹配 一個 單詞

1 2 3

發送者路由值              隊列中 old.boy.python          old. *   - -  不匹配 old.boy.python          old. #  -- 匹配

 

 1 #!/usr/bin/env python
 2 import pika
 3 import sys
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6         host='localhost'))
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange='topic_logs',
10                          type='topic')
11 
12 result = channel.queue_declare(exclusive=True)
13 queue_name = result.method.queue
14 
15 binding_keys = sys.argv[1:]
16 if not binding_keys:
17     sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])
18     sys.exit(1)
19 
20 for binding_key in binding_keys:
21     channel.queue_bind(exchange='topic_logs',
22                        queue=queue_name,
23                        routing_key=binding_key)
24 
25 print(' [*] Waiting for logs. To exit press CTRL+C')
26 
27 def callback(ch, method, properties, body):
28     print(" [x] %r:%r" % (method.routing_key, body))
29 
30 channel.basic_consume(callback,
31                       queue=queue_name,
32                       no_ack=True)
33 
34 channel.start_consuming()

消費者

 1 #!/usr/bin/env python
 2 import pika
 3 import sys
 4 
 5 connection = pika.BlockingConnection(pika.ConnectionParameters(
 6         host='localhost'))
 7 channel = connection.channel()
 8 
 9 channel.exchange_declare(exchange='topic_logs',
10                          type='topic')
11 
12 routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
13 message = ' '.join(sys.argv[2:]) or 'Hello World!'
14 channel.basic_publish(exchange='topic_logs',
15                       routing_key=routing_key,
16                       body=message)
17 print(" [x] Sent %r:%r" % (routing_key, message))
18 connection.close()

生產者

 

注意：

 

 1 sudo rabbitmqctl add_user alex 123
 2 # 設置用戶爲administrator角色
 3 sudo rabbitmqctl set_user_tags alex administrator
 4 # 設置權限
 5 sudo rabbitmqctl set_permissions -p "/" alex '.''.''.'
 6 
 7 # 而後重啓rabbiMQ服務
 8 sudo /etc/init.d/rabbitmq-server restart
 9  
10 # 而後可使用剛纔的用戶遠程鏈接rabbitmq server了。
11 
12 
13 ------------------------------
14 credentials = pika.PlainCredentials("alex","123")
15 
16 connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.14.47',credentials=credentials))

View Code

 

SQLAlchemy

SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。

Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：

1 2 3 4 5 6 7 8 9 10 11 12 13

MySQL - Python      mysql + mysqldb: / / <user>:<password>@<host>[:<port>] / <dbname>   pymysql      mysql + pymysql: / / <username>:<password>@<host> / <dbname>[?<options>]   MySQL - Connector      mysql + mysqlconnector: / / <user>:<password>@<host>[:<port>] / <dbname>   cx_Oracle      oracle + cx_oracle: / / user: pass @host:port / dbname[?key = value&key = value...]   更多詳見：http: / / docs.sqlalchemy.org / en / latest / dialects / index.html

步驟一：

使用 Engine/ConnectionPooling/Dialect 進行數據庫操做，Engine使用ConnectionPooling鏈接數據庫，而後再經過Dialect執行SQL語句。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

#!/usr/bin/env python # -*- coding:utf-8 -*-   from  sqlalchemy  import  create_engine     engine  =  create_engine( "mysql+mysqldb://root:123@127.0.0.1:3306/s11" , max_overflow = 5 )   engine.execute(      "INSERT INTO ts_test (a, b) VALUES ('2', 'v1')" )   engine.execute(       "INSERT INTO ts_test (a, b) VALUES (%s, %s)" ,      (( 555 ,  "v1" ),( 666 ,  "v1" ),) ) engine.execute(      "INSERT INTO ts_test (a, b) VALUES (%(id)s, %(name)s)" ,      id = 999 , name = "v1" )   result  =  engine.execute( 'select * from ts_test' ) result.fetchall()

 

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 
 4 from sqlalchemy import create_engine
 5 
 6 
 7 engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
 8 
 9 
10 # 事務操做
11 with engine.begin() as conn:
12     conn.execute("insert into table (x, y, z) values (1, 2, 3)")
13     conn.execute("my_special_procedure(5)")
14     
15     
16 conn = engine.connect()
17 # 事務操做 
18 with conn.begin():
19        conn.execute("some statement", {'x':5, 'y':10})

事務操做

 

注：查看數據庫鏈接：show status like 'Threads%';

步驟二：

使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 進行數據庫操做。Engine使用Schema Type建立一個特定的結構對象，以後經過SQL Expression Language將該對象轉換成SQL語句，而後經過 ConnectionPooling 鏈接數據庫，再而後經過 Dialect 執行SQL，並獲取結果。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

#!/usr/bin/env python # -*- coding:utf-8 -*-   from  sqlalchemy  import  create_engine, Table, Column, Integer, String, MetaData, ForeignKey   metadata  =  MetaData()   user  =  Table( 'user' , metadata,      Column( 'id' , Integer, primary_key = True ),      Column( 'name' , String( 20 )), )   color  =  Table( 'color' , metadata,      Column( 'id' , Integer, primary_key = True ),      Column( 'name' , String( 20 )), ) engine  =  create_engine( "mysql+mysqldb://root:123@127.0.0.1:3306/s11" , max_overflow = 5 )   metadata.create_all(engine) # metadata.clear() # metadata.remove()

 1 #!/usr/bin/env python
 2 # -*- coding:utf-8 -*-
 3 
 4 from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
 5 
 6 metadata = MetaData()
 7 
 8 user = Table('user', metadata,
 9     Column('id', Integer, primary_key=True),
10     Column('name', String(20)),
11 )
12 
13 color = Table('color', metadata,
14     Column('id', Integer, primary_key=True),
15     Column('name', String(20)),
16 )
17 engine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)
18 
19 conn = engine.connect()
20 
21 # 建立SQL語句，INSERT INTO "user" (id, name) VALUES (:id, :name)
22 conn.execute(user.insert(),{'id':7,'name':'seven'})
23 conn.close()
24 
25 # sql = user.insert().values(id=123, name='wu')
26 # conn.execute(sql)
27 # conn.close()
28 
29 # sql = user.delete().where(user.c.id > 1)
30 
31 # sql = user.update().values(fullname=user.c.name)
32 # sql = user.update().where(user.c.name == 'jack').values(name='ed')
33 
34 # sql = select([user, ])
35 # sql = select([user.c.id, ])
36 # sql = select([user.c.name, color.c.name]).where(user.c.id==color.c.id)
37 # sql = select([user.c.name]).order_by(user.c.name)
38 # sql = select([user]).group_by(user.c.name)
39 
40 # result = conn.execute(sql)
41 # print result.fetchall()
42 # conn.close()

增刪改查

 

更多內容詳見：

    http://www.jianshu.com/p/e6bba189fcbd

    http://docs.sqlalchemy.org/en/latest/core/expression_api.html

注：SQLAlchemy沒法修改表結構，若是須要可使用SQLAlchemy開發者開源的另一個軟件Alembic來完成。

步驟三：

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 全部組件對數據進行操做。根據類建立對象，對象轉換成SQL，執行SQL。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59

#!/usr/bin/env python # -*- coding:utf-8 -*-   from  sqlalchemy.ext.declarative  import  declarative_base from  sqlalchemy  import  Column, Integer, String from  sqlalchemy.orm  import  sessionmaker from  sqlalchemy  import  create_engine   engine  =  create_engine( "mysql+mysqldb://root:123@127.0.0.1:3306/s11" , max_overflow = 5 )   Base  =  declarative_base()     class  User(Base):      __tablename__  =  'users'      id  =  Column(Integer, primary_key = True )      name  =  Column(String( 50 ))   # 尋找Base的全部子類，按照子類的結構在數據庫中生成對應的數據表信息 # Base.metadata.create_all(engine)   Session  =  sessionmaker(bind = engine) session  =  Session()     # ########## 增 ########## # u = User(id=2, name='sb') # session.add(u) # session.add_all([ #     User(id=3, name='sb'), #     User(id=4, name='sb') # ]) # session.commit()   # ########## 刪除 ########## # session.query(User).filter(User.id > 2).delete() # session.commit()   # ########## 修改 ########## # session.query(User).filter(User.id > 2).update({'cluster_id' : 0}) # session.commit() # ########## 查 ########## # ret = session.query(User).filter_by(name='sb').first()   # ret = session.query(User).filter_by(name='sb').all() # print ret   # ret = session.query(User).filter(User.name.in_(['sb','bb'])).all() # print ret   # ret = session.query(User.name.label('name_label')).all() # print ret,type(ret)   # ret = session.query(User).order_by(User.id).all() # print ret   # ret = session.query(User).order_by(User.id)[1:3] # print ret # session.commit()

更多功能參見文檔，猛擊這裏下載PDF