Python操作 RabbitMQ、Redis、Memcache、SQLAlchemy
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、第一次操作
?| 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 表示運行出現錯誤時，現實錯誤信息，上線后移除該參數。
2、天生支持集群
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') |
3、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重復添加，失敗！！！ |
4、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') |
5、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'}) |
6、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']) |
7、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"]) |
8、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" |
9、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 |
10、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(主從)同步。
一、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" |
二、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 操作
- 管道
- 發布訂閱
?
1、操作模式
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') |
2、連接池
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') |
3、操作
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的二進制表示為：011001100110111101101111????所以，如果執行 setbit('n1',7,1)，則就會將第7位設置為1，????????那么最終二進制則變成011001110110111101101111，即："goo"# 擴展，轉換二進制表示：????# source = "武沛齊"????source="foo"????foriinsource:????????num=ord(i)????????bin(num).replace('b','')????特別的，如果source是漢字"武沛齊"怎么辦？????答：對于utf-8，每一個漢字占3個字節，那么"武沛齊"則有9個字節???????對于漢字，for循環時候會按照 字節 迭代，那么在迭代時，將每一個字節轉換 十進制數，然后再將十進制數轉換成二進制????????111001101010110110100110111001101011001010011011111010011011110110010000????????------------------------------------------------------------------------------------????????????????????武???????????????????????? 沛?????????????????????????? 齊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,自增數（必須是整數）# 注：同incrbyincrbyfloat(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獲取valuehmget(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是否存在當前傳入的keyhdel(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=amounthscan(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，要添加數據的列表的nameblpop(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對應的列表的所有元素，那么就需要：????# 1、獲取name對應的所有列表????# 2、循環列表# 但是，如果列表非常大，那么就有可能在第一步時就將程序的內容撐爆，所有有必要自定義一個增量迭代的功能：deflist_iter(name):????"""????自定義redis列表增量迭代????:param name: redis中的name，即：迭代name對應的列表????:return: yield 返回 列表元素????"""????list_count=r.llen(name)????forindexinxrange(list_count):????????yieldr.lindex(name, index)# 使用foriteminlist_iter('pp'):????itemSet操作，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? MAXzunionstore(dest, keys, aggregate=None)
?
1 2 # 獲取兩個有序集合的并集，如果遇到相同值不同分數，則按照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)
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 ，但不匹配 hilloexpire(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
4、管道
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) r.set('name', 'alex') r.set('role', 'sb') pipe.execute() |
5、發布訂閱
發布者：服務器
訂閱者：Dashboad和數據處理
Demo如下：
#!/usr/bin/env python # -*- coding:utf-8 -*-import redisclass 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 Truedef subscribe(self):pub = self.__conn.pubsub()pub.subscribe(self.chan_sub)pub.parse_response()return pubRedisHelper
訂閱者：
?| 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實現生產者消費者模型
#!/usr/bin/env python # -*- coding:utf-8 -*- import Queue import threadingmessage = 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來說，生產和消費不再針對內存里的一個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() |
1、acknowledgment 消息不丟失
no-ack ＝ False，如果消費者遇到情況(its channel is closed, connection is closed, or TCP connection is lost)掛掉了，那么，RabbitMQ會重新將該任務添加到隊列中。
import pikaconnection = 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 timetime.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 ? 消息不丟失
#!/usr/bin/env python import pikaconnection = 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 pikaconnection = 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 timetime.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 去隊列中獲取 奇數 序列的任務，消費者1去隊列中獲取 偶數 序列的任務。
channel.basic_qos(prefetch_count=1) 表示誰來誰取，不再按照奇偶數排列
#!/usr/bin/env python # -*- coding:utf-8 -*- import pikaconnection = 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 timetime.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 sysconnection = 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 pikaconnection = 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.queuechannel.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、關鍵字發送
?exchange type = direct
之前事例，發送消息時明確指定某個隊列并向其中發送消息，RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 判定應該將數據發送至指定隊列。
#!/usr/bin/env python import pika import sysconnection = 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.queueseverities = 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 sysconnection = 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 個 或 多個 單詞
- * ?表示只能匹配 一個 單詞
| 1 2 3 | 發送者路由值????????????? 隊列中 old.boy.python????????? old.*? -- 不匹配 old.boy.python????????? old.#? -- 匹配 |
#!/usr/bin/env python import pika import sysconnection = 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.queuebinding_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 sysconnection = 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()生產者
?
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() |
#!/usr/bin/env python # -*- coding:utf-8 -*-from sqlalchemy import create_engineengine = create_engine("mysql+mysqldb://root:123@127.0.0.1:3306/s11", max_overflow=5)# 事務操作 with engine.begin() as conn:conn.execute("insert into table (x, y, z) values (1, 2, 3)")conn.execute("my_special_procedure(5)")conn = engine.connect() # 事務操作 with conn.begin():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() |
#!/usr/bin/env python # -*- coding:utf-8 -*-from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKeymetadata = 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)conn = engine.connect()# 創建SQL語句，INSERT INTO "user" (id, name) VALUES (:id, :name) conn.execute(user.insert(),{'id':7,'name':'seven'}) conn.close()# sql = user.insert().values(id=123, name='wu') # conn.execute(sql) # conn.close()# sql = user.delete().where(user.c.id > 1)# sql = user.update().values(fullname=user.c.name) # sql = user.update().where(user.c.name == 'jack').values(name='ed')# sql = select([user, ]) # sql = select([user.c.id, ]) # sql = select([user.c.name, color.c.name]).where(user.c.id==color.c.id) # sql = select([user.c.name]).order_by(user.c.name) # sql = select([user]).group_by(user.c.name)# result = conn.execute(sql) # print result.fetchall() # 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
分類: python基礎到高級 好文要頂 關注我 收藏該文 wangheng1409關注 - 15
粉絲 - 137 +加關注 2 0 ? 上一篇：Python操作MySQL
? 下一篇：tornado web高級開發項目之抽屜官網的頁面登陸驗證、form驗證、點贊、評論、文章分頁處理、發送郵箱驗證碼、登陸驗證碼、注冊、發布文章、上傳圖片
轉載于:https://www.cnblogs.com/ExMan/p/10438842.html
總結
以上是生活随笔為你收集整理的Python操作 RabbitMQ、Redis、Memcache、SQLAlchemy的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 房屋装修多少钱啊？
- 下一篇： 求一个体智能老师好听的名字！