四. python網絡編程
第八章.網絡基礎知識
1. TCP/IP協議介紹
1.TCP/IP概念html
TCP/IP: Transmission Control Protocol/Internet Protocol的簡寫，中譯名爲傳輸控制協議/因特網互聯協議，又名網絡通信協議，是Internet最基本的協議、Internet國際互聯網絡的基礎，由網絡層的IP協議和傳輸層的TCP協議組成。TCP/IP 定義了電子設備如何連入因特網，以及數據如何在它們之間傳輸的標準。協議採用了4層的層級結構，每一層都呼叫它的下一層所提供的協議來完成本身的需求。通俗而言：TCP負責發現傳輸的問題，一有問題就發出信號，要求從新傳輸，直到全部數據安全正確地傳輸到目的地。而IP是給因特網的每一臺聯網設備規定一個地址。 java
2. socket概念python
socket一般也稱做"套接字"，用於描述IP地址和端口，是一個通訊鏈的句柄，應用程序一般經過"套接字"向網絡發出請求或者應答網絡請求。mysql
socket起源於Unix，而Unix/Linux基本哲學之一就是「一切皆文件」，對於文件用【打開】【讀寫】【關閉】模式來操做。socket就是該模式的一個實現，socket便是一種特殊的文件，一些socket函數就是對其進行的操做（讀/寫IO、打開、關閉）linux
socket和file的區別：git
- file模塊是針對某個指定文件進行【打開】【讀寫】【關閉】
- socket模塊是針對 服務器端 和 客戶端Socket 進行【打開】【讀寫】【關閉】
3. OSI七層模型與 TCP/IP協議分層圖github
2.socket編程
1. socke通訊模型圖面試
2. 基本語法類型redis
例1算法
# -*- coding:utf-8 -*- # Author: Raymond import socket server = socket.socket() server.bind(('localhost',6969)) #綁定要監聽的端口 server.listen() #監聽上面bind綁定的端口 print("我要開始等電話了....") conn,addr = server.accept() #等待電話打進來 print("電話來了") data = conn.recv(1024) print("服務端收到的信息",data) conn.send(data.upper()) #或修改成conn.send(b"hello client!") server.close()
# -*- coding:utf-8 -*- # Author: Raymond import socket client = socket.socket() #聲明socket類型，同時生成socket鏈接對象 client.connect(('localhost',6969)) client.send(b"hello world!") #或修改成client.send(b"hello server!") data = client.recv(1024) print("客戶端收到的信息:",data) client.close()
例2
# server code import socket ip_port = ('localhost', 9999) s = socket.socket() # 買手機 s.bind(ip_port) # 買手機卡 s.listen(5) # 開機，5個連接數掛起排隊 conn,addr = s.accept() #等待電話；conn表示一個客戶端與服務器的一條通訊鏈路 while True: try: recv_data = conn.recv(1024) # 收消息，客戶端發空消息，服務端recv會被阻塞，沒法接受消息 if str(recv_data,encoding='utf-8') == 'exit': break print(str(recv_data,encoding='utf-8'),type(recv_data)) send_data = recv_data.upper() # 發消息 conn.send(send_data) except Exception: break conn.close() # 掛電話
# client code import socket ip_port = ('localhost',9999) s = socket.socket() # 買手機 s.connect(ip_port) # 撥號 while True: send_data = input(">>:").strip() # 輸入消息內容 if len(send_data) == 0: continue s.send(bytes(send_data,encoding='utf-8')) # 發消息,python 3.x 必須以字節碼發送 if send_data == 'exit': break recv_data = s.recv(1024) # 收消息 print(str(recv_data,encoding='utf-8')) # 打印bytes字節碼 s.close() # 掛電話 客戶端示例(二)
3. socket經常使用方法
更多功能
| 參數一：地址簇 socket.AF_INET IPv4（默認） socket.AF_UNIX 只可以用於單一的Unix系統進程間通訊 參數二：類型 socket.SOCK_STREAM 流式socket , for TCP （默認） socket.SOCK_RAW 原始套接字，普通的套接字沒法處理ICMP、IGMP等網絡報文，而SOCK_RAW能夠；其次，SOCK_RAW也能夠處理特殊的IPv4報文；此外，利用原始套接字，能夠經過IP_HDRINCL套接字選項由用戶構造IP頭。 參數三：協議 0 （默認）與特定的地址家族相關的協議,若是是 0 ，則系統就會根據地址格式和套接類別,自動選擇一個合適的協議
import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) sk.bind(ip_port) while True: data = sk.recv(1024) print data import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) while True: inp = raw_input('數據：').strip() if inp == 'exit': break sk.sendto(inp,ip_port) sk.close() |
sk.bind(address)
s.bind(address) 將套接字綁定到地址。address地址的格式取決於地址族。在AF_INET下，以元組（host,port）的形式表示地址。
sk.listen(backlog)
開始監聽傳入鏈接。backlog指定在拒絕鏈接以前，能夠掛起的最大鏈接數量。
backlog等於5，表示內核已經接到了鏈接請求，但服務器尚未調用accept進行處理的鏈接個數最大爲5
這個值不能無限大，由於要在內核中維護鏈接隊列
sk.setblocking(bool)
是否阻塞（默認True），若是設置False，那麼accept和recv時一旦無數據，則報錯。
sk.accept()
接受鏈接並返回（conn,address）,其中conn是新的套接字對象，能夠用來接收和發送數據。address是鏈接客戶端的地址。
接收TCP 客戶的鏈接（阻塞式）等待鏈接的到來
sk.connect(address)
鏈接到address處的套接字。通常，address的格式爲元組（hostname,port）,若是鏈接出錯，返回socket.error錯誤。
sk.connect_ex(address)
同上，只不過會有返回值，鏈接成功時返回 0 ，鏈接失敗時候返回編碼，例如：10061
sk.close()
關閉套接字
sk.recv(bufsize[,flag])
接受套接字的數據。數據以字符串形式返回，bufsize指定最多能夠接收的數量。flag提供有關消息的其餘信息，一般能夠忽略。
sk.recvfrom(bufsize[.flag])
與recv()相似，但返回值是（data,address）。其中data是包含接收數據的字符串，address是發送數據的套接字地址。
sk.send(string[,flag])
將string中的數據發送到鏈接的套接字。返回值是要發送的字節數量，該數量可能小於string的字節大小。即：可能未將指定內容所有發送。
sk.sendall(string[,flag])
將string中的數據發送到鏈接的套接字，但在返回以前會嘗試發送全部數據。成功返回None，失敗則拋出異常。
內部經過遞歸調用send，將全部內容發送出去。
sk.sendto(string[,flag],address)
將數據發送到套接字，address是形式爲（ipaddr，port）的元組，指定遠程地址。返回值是發送的字節數。該函數主要用於UDP協議。
sk.settimeout(timeout)
設置套接字操做的超時期，timeout是一個浮點數，單位是秒。值爲None表示沒有超時期。通常，超時期應該在剛建立套接字時設置，由於它們可能用於鏈接的操做（如 client 鏈接最多等待5s ）
sk.getpeername()
返回鏈接套接字的遠程地址。返回值一般是元組（ipaddr,port）。
sk.getsockname()
返回套接字本身的地址。一般是一個元組(ipaddr,port)
sk.fileno()
套接字的文件描述符
# 服務端 import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) sk.bind(ip_port) while True: data,(host,port) = sk.recvfrom(1024) print(data,host,port) sk.sendto(bytes('ok', encoding='utf-8'), (host,port)) #客戶端 import socket ip_port = ('127.0.0.1',9999) sk = socket.socket(socket.AF_INET,socket.SOCK_DGRAM,0) while True: inp = input('數據：').strip() if inp == 'exit': break sk.sendto(bytes(inp, encoding='utf-8'),ip_port) data = sk.recvfrom(1024) print(data) sk.close()
實例：智能機器人
#!/usr/bin/env python # -*- coding:utf-8 -*- import socket ip_port = ('127.0.0.1',8888) sk = socket.socket() sk.bind(ip_port) sk.listen(5) while True: conn,address = sk.accept() conn.sendall('歡迎致電 10086，請輸入1xxx,0轉人工服務.') Flag = True while Flag: data = conn.recv(1024) if data == 'exit': Flag = False elif data == '0': conn.sendall('經過可能會被錄音.balabala一大推') else: conn.sendall('請從新輸入.') conn.close()
#!/usr/bin/env python # -*- coding:utf-8 -*- import socket ip_port = ('127.0.0.1',8005) sk = socket.socket() sk.connect(ip_port) sk.settimeout(5) while True: data = sk.recv(1024) print 'receive:',data inp = raw_input('please input:') sk.sendall(inp) if inp == 'exit': break sk.close()
3. socket簡單ssh交互
- 基於python 3.x版本的socket智能手法字節碼，python 2.x能夠收發字符串
- socket中循環輸入功能只在客戶端退出就能夠了。
- socket編程中accept()和recv()是阻塞的。(基於連接正常)
- listen(n) n表明：能掛起的連接數；若是n = 1,表明能夠連接一個，掛起一個，第三個拒絕
- 服務端出現端口衝突，修改監聽端口號
- 以太網協議傳送最大的包爲1500字節。(工業標準1024-8192字節)
import socket import subprocess ip_port = ('localhost', 9999) s = socket.socket() # 買手機 s.bind(ip_port) # 買手機卡 s.listen(5) # 開機，5個連接數掛起排隊 conn,addr = s.accept() # 等待電話；conn表示一個客戶端與服務器的一條通訊鏈路 while True: try: recv_data = conn.recv(1024) # 收消息，客戶端發空消息，服務端recv會被阻塞，沒法接受消息 if str(recv_data,encoding='utf-8') == 'exit': break print(str(recv_data,encoding='utf-8'),type(recv_data)) p = subprocess.Popen(str(recv_data,encoding='utf-8'),shell=True,stdout=subprocess.PIPE) res = p.stdout.read() # 返回系統命令的標準輸出結果 send_data = str(res,encoding='gbk') print(send_data) # send_data = recv_data.upper() # 發消息 conn.send(bytes(send_data,encoding='utf-8')) except Exception: break conn.close() # 掛電話
import socket ip_port = ('localhost',9999) s = socket.socket() # 買手機 s.connect(ip_port) # 撥號 while True: send_data = input(">>:").strip() # 輸入消息內容 if len(send_data) == 0: continue s.send(bytes(send_data,encoding='utf-8')) # 發消息,python 3.x 必須以字節碼發送 if send_data == 'exit': break recv_data = s.recv(1024) # 收消息 print(str(recv_data,encoding='utf-8')) # 打印bytes字節碼 s.close() # 掛電話
4. socket粘包解決
粘包概念：客戶端發發一條命令，服務端返回一部分結果，剩餘部分丟失，丟失部分會出如今下一條命令返回的結果中
# 粘包問題 需調試 import socket import subprocess ip_port = ('localhost', 9999) s = socket.socket() s.bind(ip_port) s.listen(5) while True: # 用來重複接收新的連接 conn, addr = s.accept() # 接收客戶端連接請求，返回conn(至關於一個特定連接)，addr是客戶端ip+port while True: # 用來基於一個連接重複收發消息 try: # 捕捉客戶端異常關閉錯誤 recv_data = conn.recv(1024) # 收消息，阻塞 if len(recv_data) == 0:break # 客戶端退出，服務端將收到空消息 p = subprocess.Popen(str(recv_data,encoding='utf-8'),shell=True,stdout=subprocess.PIPE) # 執行系統命令 res = p.stdout.read() if len(res) == 0: send_data = 'cmd_error' else: send_data = str(res,encoding='gbk') # windows平臺命令的標準輸出爲GBK編碼，須要轉換 send_data = bytes(send_data,encoding='utf-8') # 解決粘包問題 ready_tag = 'Ready|%s' %len(send_data) conn.send(bytes(ready_tag,encoding='utf-8')) feedback = conn.recv(1024) # start feedback = str(feedback,encoding='utf-8') if feedback.startswith('Start'): conn.send(send_data) # 發送命令的執行結果 except Exception: break conn.close()
# 粘包問題 import socket ip_port = ('localhost',9999) s = socket.socket() s.connect(ip_port) while True: send_data = input(">>:").strip() if len(send_data) == 0:continue s.send(bytes(send_data,encoding='utf-8')) if send_data == 'exit':break # 解決粘包問題 ready_tag = s.recv(1024) ready_tag = str(ready_tag,encoding='utf-8') if ready_tag.startswith('Ready'): msg_size = int(ready_tag.split('|')[-1]) start_tag = 'Start' s.send(bytes(start_tag,encoding='utf-8')) recv_size = 0 recv_msg = b'' while recv_size < msg_size: recv_data = s.recv(1024) recv_msg += recv_data recv_size += len(recv_data) print('MSG SIZE %s RECEV SIZE %s' %(msg_size,recv_size)) # 收消息 print(str(recv_data,encoding='utf-8')) s.close()
5. socketServer介紹
6. 實現多併發命令執行
7. 實現文件上傳下載
8. python吊炸天的小知識之做用域
- python中無塊級做用域，代碼塊中定義的變量可在任意地方使用
- python中以函數爲做用域，函數中的變量不可在函數外使用
- python中有做用域鏈，由內向外找，直到找不到報錯
- 在函數未執行前，做用域以及鏈已肯定
小知識之做用域 ############################################################################# # 代碼塊 if 1==1: name = 'alex' print(name) 輸出結果：alex ############################################################################ # 函數 def func(): name = 'alex' func() print(name) 輸出結果：alex ############################################################################# # 做用域一 name = 'alex' def f1(): name = 'a' def f2(): name = 'b' print(name) f2() f1() 輸出結果：b ############################################################################# # 做用域二 name = 'alex' def f1(): print(name) def f2(): name = 'eric' f1() f2() 輸出結果：alex ############################################################################# # 做用域三 name = 'alex' def f1(): print(name) def f2(): name = 'eric' return f1 ret = f2() ret() 輸出結果：alex
9. python吊炸天的小知識之面試題
#小知識之面試題 ############################################################################# li = [x+100 for x in range(10)] print(li) 輸出結果：[100, 101, 102, 103, 104, 105, 106, 107, 108, 109] #------------------------------------------------------------------------------- li = [x+100 for x in range(10) if x>6] print(li) 輸出結果：[107, 108, 109] ------------------------------------------------------------------------------- li = [lambda :x for x in range(10)] r = li[0]() print(r) 輸出結果：9 ------------------------------------------------------------------------------- li = [] for i in range(10): def f1(): return i li.append(f1) # # li是列表，內部元素是相同功能的函數 # print(i) print(li[0]()) print(li[1]()) 輸出結果; 9 9 ------------------------------------------------------------------------------- li = [] for i in range(10): def f1(x=i): # 本質看是否執行此行代碼，執行了就能夠取到值，不執行就取不到值。本行執行了。 return x li.append(f1) # # li是列表，內部元素是相同功能的函數 # print(i) print(li[0]()) print(li[1]()) print(li[2]()) 輸出結果： 0 1 2
10. I/O多路複用實現僞併發
I/O多路複用：監聽socket內部對象是否發生變化
# I/O多路複用概述：用來監聽socket對象內部是否變化了 import socket import select sk = socket.socket() sk.bind(('127.0.0.1',9999)) sk.listen(5) while True: rlist,w,e = select.select([sk],[],[],1) # select()內部監聽socket對象，循環等1秒；無鏈接，一直循環下去; print(rlist) # sk,sk1,sk2其中哪一個變化，則r返回哪一個值 # r = [sk,] # r獲取socket對象列表，可判斷是否有新鏈接 # r = [sk1,] # r = [sk1,sk2] # 這兩個同時鏈接，返回兩個值 for r in rlist: print(r) # r獲取socket對象列表，可判斷是否有新連接 conn,addrs = r.accept() # 接收客戶端連接，Conn也是socket對象 conn.sendall(bytes('hello',encoding='utf-8')) 無限循環輸出結果： [<socket.socket fd=396, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999)>] <socket.socket fd=396, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9999)>
import socket sk = socket.socket() sk.connect(('127.0.0.1',9999,)) data = sk.recv(1024) print(data) while True: input('>>>') sk.close()
I/O多路複用：僞併發實例
# I/O多路複用僞併發實例 import socket import select sk = socket.socket() sk.bind(('127.0.0.1',9999)) sk.listen(5) inputs = [sk,] while True: rlist,w,e = select.select([sk],[],[],1) print(len(inputs),len(rlist)) for r in rlist: if r == sk: # 新客戶端來鏈接 print(r) conn,addrs = r.accept() inputs.append(conn) # 此處將新鏈接客戶端添加到inputs列表中 conn.sendall(bytes('hello', encoding='utf-8')) else: r.recv(1024)
import socket sk = socket.socket() sk.connect(('127.0.0.1',9999,)) data = sk.recv(1024) print(data) while True: input('>>>') sk.close()
I/O多路複用：鏈接的客戶端斷開後，服務端進行拋出異常處理，並移除斷開的客戶端
# 鏈接的客戶端斷開後，服務端進行拋出異常處理，並移除斷開的客戶端 import socket import select sk = socket.socket() sk.bind(('127.0.0.1',9999)) sk.listen(5) inputs = [sk,] while True: rlist,w,e = select.select([sk],[],[],1) print(len(inputs),len(rlist)) for r in rlist: if r == sk: # 新客戶端來鏈接 print(r) conn,addrs = r.accept() inputs.append(conn) # 此處將新鏈接客戶端添加到inputs列表中 conn.sendall(bytes('hello', encoding='utf-8')) else: print('==============') # 有人給我發消息了 try: ret = r.recv(1024) if not ret: raise Exception('斷開鏈接') # 主動拋出異常 except Exception as e: inputs.remove(r) # 若是客戶端突發斷開鏈接，此處則移除斷掉的客戶端
import socket sk = socket.socket() sk.connect(('127.0.0.1',9999,)) data = sk.recv(1024) print(data) while True: input('>>>') sk.close()
I/O多路複用：讀寫分離(客戶端發消息，服務端回發消息)
# I/O多路複用之讀寫分離(客戶端發消息，服務端回發消息) import socket import select sk = socket.socket() sk.bind(('127.0.0.1',9999)) sk.listen(5) inputs = [sk,] outputs = [] while True: rlist,wlist,e = select.select(inputs,outputs,[],1) # outputs表示全部客戶端給服務端發的消息，outputs==wlist print(len(inputs),len(rlist),len(outputs),len(wlist)) for r in rlist: if r == sk: # 新客戶端來鏈接 print(r) conn,addrs = r.accept() inputs.append(conn) # 此處將新鏈接客戶端添加到inputs列表中 conn.sendall(bytes('hello', encoding='utf-8')) else: print('==============') # 有人給我發消息了 try: ret = r.recv(1024) r.sendall(ret) if not ret: raise Exception('斷開鏈接') # 主動拋出異常 else: outputs.append(r) except Exception as e: inputs.remove(r) # 若是客戶端突發斷開鏈接，此處則移除斷掉的客戶端 for w in wlist: # 全部給我發消息的客戶端 w.sendall(bytes('response',encoding='utf-8')) outputs.remove(w)
import socket sk = socket.socket() sk.connect(('127.0.0.1',9998,)) data = sk.recv(1024) print(data) while True: inp = input('>>>') sk.sendall(bytes(inp,encoding='utf-8')) # 客戶端發消息，服務端回發消息 print(sk.recv(1024)) sk.close() View Code
I/O多路複用：讀寫分離(客戶端發一條消息，服務端回發：客戶端消息+服務端消息)
# I/O多路複用之讀寫分離(客戶端發一條消息，服務端回發：客戶端消息+服務端消息) 需調試！！！ import socket import select sk = socket.socket() sk.bind(('127.0.0.1',9998)) sk.listen(5) inputs = [sk,] outputs = [] message = {} ''' message = { 張三:[msg1,msg2] 李四:[msg1,msg2] } ''' while True: rlist,wlist,elist = select.select(inputs,outputs,[sk,],1) # outputs表示全部客戶端給服務端發的消息，outputs==wlist;第三個參數sk也是socket對象，內部會檢測sk內部發生錯誤，會把sk賦值給elist print(len(inputs),len(rlist),len(outputs),len(wlist)) for r in rlist: if r == sk: # 新客戶端來鏈接 print(r) conn,addrs = r.accept() inputs.append(conn) # 此處將新鏈接客戶端添加到inputs列表中 message[conn] = [] # 客戶端來鏈接則把conn對象看成key值放入消息字典中，而後在存入空列表中 conn.sendall(bytes('hello', encoding='utf-8')) else: print('==============') # 有人給我發消息了 try: ret = r.recv(1024) r.sendall(ret) if not ret: raise Exception('斷開鏈接') # 主動拋出異常 else: outputs.append(r) message[r].append(ret) # 把客戶端發的消息放到ret列表裏 except Exception as e: inputs.remove(r) # 若是客戶端突發斷開鏈接，此處則移除斷掉的客戶端 del message[r] for w in wlist: # 全部給我發消息的客戶端 msg = message[w].pop() resp = msg + bytes('response',encoding='utf-8') w.sendall(resp) outputs.remove(w)
import socket sk = socket.socket() sk.connect(('127.0.0.1',9998,)) data = sk.recv(1024) print(data) while True: inp = input('>>>') sk.sendall(bytes(inp,encoding='utf-8')) # 客戶端發消息，服務端回發消息 print(sk.recv(1024)) sk.close()
11. 初識多線程
12. socket源碼剖析
13. 線程進程概述
- 一個python應用程序，能夠有多個進程和多個線程
- 默認爲單進程和單線程(效率低)
- 多線程上面有一個GIL(全局解釋器鎖)，給進程起到屏障做用，進程裏的多線程同一時間只能有一個線程工做，不能同時被CPU調度工做。(java,c#裏沒有GIL功能)
- python多線程應用場景：1. I/O操做不佔用CPU(音頻，視頻，顯卡都交給驅動去工做)，適合用多線程提升併發效率；2. 計算型操做佔用CPU，適合用多進程提升併發效率，每一個進程可利用多核CPU同時進行並行計算。
- I/O密集型操做：用多線程； 計算型操做：用多進程
進程、線程與GIL關係原理圖以下所示：
14. 線程基本使用
# 線程基本使用
import time
def f1(arg): # f1()表示一個單進程，單線程應用程序
time.sleep(5)
print(arg) # f1()默認爲主線程，t表示新建立的子線程
# f1表示主線程，下面的t表示建立的子線程,建立好線程，等待CPU調度工做，不表明當前線程會被當即執行
import threading # 導入線程模塊
t = threading.Thread(target=f1,args=(123,)) # 此處建立一個子線程，target=f1表示執行f1函數，並傳參數args給f1(arg)
# t.setDaemon(True) # 參數爲True時，表示主線程不等待子線程執行完畢直接輸出end；則整個程序執行完畢；False則表示主線程執行完後，繼續等待子線程執行完畢
t.start()
t.join(2) # 表示主線程執行到此處，等待子線程執行完畢後，再繼續執行下面的代碼；參數(2)表示主線程運行到此處最多等待2秒
print('end')
# 仔細閱讀socketserver源碼(與其餘語言原理同樣)
# 仔細閱讀saltstack源碼 ,提升能力
第九章.線程、進程、協程和緩存的基本使用
1. 本章內容概要
- 線程：1.基本使用；2.生產者消費者模型(隊列)；3自定義線程池(2.x沒有此概念；3.x還不完善)
- 進程：1.基本使用；2.自己提供的進程池
- 協程：兩個模塊 greenlet和gevent
- 緩存：python操做的兩個緩存 memcached和redis
- rabbitMQ
- MySQL
2. 建立線程的兩種方式
# 建立線程的兩種方式 #第一種 import threading def f1(arg): print(arg) t = threading.Thread(target=f1,args=(123,)) t.start() # 建立好線程，等待CPU調用 t.run() # CPU調度線程工做時，會去執行t.run()方法，該方法再去觸發線程工做 # ------------------------------------------------------------------------------- # 自定義類建立線程並去執行 # 第二種方法 import threading class MyThread(threading.Thread): def __init__(self,func,args): # 當前類的構造方法 self.func = func self.args = args super(MyThread,self).__init__() # 執行父類的構造方法 def run(self): # run()方法裏可實現任意功能 self.func(self.args) def f2(): print(arg) obj = MyThread(f2,123) obj.start()
3. 先進先出隊列的使用
import queue q = queue.Queue(10) # 建立一個先進先出隊列，參數表示最多放10個數據 q.put(11) # 依次把數據11，22，33放入隊列 q.put(22) q.put(33) print(q.qsize()) # 輸出隊列元素個數，檢測隊列真實個數 print(q.get()) # 從隊列取數據 print(q.get()) print(q.get()) 輸出結果： 3 11 22 33 ----------------------------------------------------------------------------------------- import queue q = queue.Queue(10) # 建立一個先進先出隊列，參數表示最多放10個數據 q.put(11) # 依次把數據11，22，33放入隊列 q.put(22) q.put(23) q.put(24) q.put(25) print(q.qsize()) # 輸出隊列元素個數，檢測隊列真實個數 q.put(33,timeout=2) # timeout表示超時時間，等2秒若是隊列有位置加入，無位置則報錯 q.put(34,block=False) # 此處再也不阻塞，直接報錯 print(q.get()) # 從隊列取數據 print(q.get()) print(q.get()) # print(q.get(block=False)) # 再也不阻塞等待取數據，直接報錯 # ----------------------------------------------------------------------------------------- import queue q = queue.Queue() q.put(123) q.put(456) q.get() q.get() q.task_done() # 去除元素後，告訴隊列當前任務已完成 # q.join() # 表示若是隊列任務沒有完成，則等待 print('end') # task_done()與join() 通常聯合使用，當隊列任務執行完畢，再也不阻塞
4. 其餘隊列介紹
queue.Queue() # 先進先出隊列
queue.LifoQueue() # 後進先出隊列
queue.PriorityQueue() # 優先級隊列
queue.deque() #雙向隊列
# 後進先出隊列 import queue q = queue.LifoQueue() q.put(123) q.put(456) print(q.get()) print(q.get()) 輸出結果： 456 123
# 優先級隊列 (0-n,優先級由高到低) import queue q = queue.PriorityQueue() q.put((1,'a')) q.put((0,'b')) q.put((2,'c')) print(q.get()) 輸出結果：(0, 'b')
# 雙向隊列 import queue q = queue.deque() q.append(123) # 放數據 q.append(456) q.appendleft(789) # 左邊放數據 print(q.pop()) # 取數據 print(q.popleft()) # 左邊取數據 輸出結果： 456 789
5. 生產者消費者模型
# 生產消費者模型 import queue import threading import time q = queue.Queue() # 建立空隊列 def productor(arg): # 買票訂單處理，加入隊列 q.put(str(arg)+'-票') def consumer(arg) # 服務器後臺賣票處理請求 while True: print(arg,q.get()) time.sleep(2) # 等待2秒後繼續處理 for i in range(300): # 建立新的300個買票者 t = threading.Thread(target=productor,args=(i,)) t.start() for j in range(3): # 建立3個賣票者循環處理請求，每次可併發處理3個請求， t = threading.Thread(target=consumer,args=(j,)) t.start()
6. 線程鎖
# 線程鎖 import threading import time NUM = 10 def func(lo): global NUM lo.acquire() # 上鎖，表示同一時刻只容許一個線程來操做 NUM -= 1 time.sleep(2) print(NUM) lo.release() # 開鎖 # lock = threading.Lock() # 單層鎖 lock = threading.RLock() # 多層鎖 for i in range(10): t = threading.Thread(target=func,args=(lock,)) t.start() View Code
7. 信號量以及事件
# 信號量以及事件 # 事件：表示把線程所有鎖上，再一次性放開工做，至關於紅綠燈，線程是車輛 import threading def func(i,e): print(i) e.wait() # 所有線程中止到這裏，這裏檢測是什麼燈；默認是紅燈 print(i+100) event = threading.Event() for i in range(10): t = threading.Thread(target=func,args=(i,event)) t.start() event.clear() # 表示所有中止，設置成紅燈 inp = input('>>>:') if inp == "1": event.set() # 表示把上面中止的線程放行，設置成綠燈 輸出結果： 0 1 2 3 4 5 6 7 8 9 >>>:1 102 104 109 106 105 100 103 107 101 108
8. 條件以及定時器
# 條件以及定時器 # 第一種條件鎖,線程等待輸入條件，才能調用工做 import threading def condition(): ret = False r = input('>>>') if r == 'true': ret = True else: ret = False return ret def func(i,con): print(i) con.acquire() con.wait() print(i+100) con.release() c = threading.Condition() for i in range(10): t = threading.Thread(target=func,args=(i,c,)) t.start() while True: # 主線程 inp = input('>>>') # 輸入幾，就放出幾個線程 if inp == 'q': break c.acquire() c.notify(int(inp)) c.release() 輸出結果： 0 1 2 3 4 5 6 7 8 9 >>>1 >>>100 2 >>>101 102 3 >>>103 105 104 # ########################################################################################### # 當知足某個條件時，才讓線程一個一個出去 import threading def condition(): ret = False r = input('>>>') if r == 'true': ret = True else: ret = False return ret def func(i,con): print(i) con.acquire() con.wait_for(condition) # 等待某個條件成立，才執行。不然一直等待 print(i+100) con.release() c = threading.Condition() for i in range(10): t = threading.Thread(target=func,args=(i,c,)) t.start() 輸出結果： 0 >>>1 2 3 4 5 6 7 8 9 true >>>101 >>>true 102 >>>false >>>
# 定時器 from threading import Timer def hello(): print("hello,world") t = Timer(2,hello) # 延遲2秒後執行hello()函數 t.start() print('end') # 此處當即打印後，再延遲2秒輸出上面函數結果
9. 自定義線程池
- 設置一個容器(有最大個數限制)，建立隊列放線程；
- 添加令牌，取一個，少一個；
- 無線程時等待
- 線程執行完畢，交還線程；
# 流程：1.定義一個容器(最大個數)；2.取一個，少一個；3.無線程時等待；4. 線程執行完畢，交還線程； # # 自定義線程池,每5個線程執行一次 import threading import queue import time class Threadpool(): def __init__(self,maxsize=5): # 建立線程池，maxsize表示線程池最多有多少個線程 self.maxsize = maxsize self._q = queue.Queue(maxsize) # 隊列此刻爲空，最多放了5個元素，元素爲類 for i in range(maxsize): self._q.put(threading.Thread) # 表示把一個類名逐個放入線程池 def get_thread(self): return self._q.get() # 獲取線程，取一個線程類，上面線程池少一個 def add_thread(self): self._q.put(threading.Thread) # 線程少一個時，又增長一個 pool = Threadpool(5) # 建立線程池，最大個數爲5個線程可使用 def task(arg,p): # 獲取pool傳遞的參數， print(arg) time.sleep(1) p.add_thread() # 表示任務執行完，再增長一個，這樣會每5個任務併發執行一次 for i in range(100): # 建立100個任務 t = pool.get_thread() # 從線程池獲取一個線程類 obj = t(target=task,args=(i,pool)) # 獲取一個線程對象，加pool參數，傳遞到task函數的參數p. obj.start() # 執行
10. 進程基本操做
# 進程基本操做 # 多進程適用於linux from multiprocessing import Process from multiprocessing import Array from multiprocessing import Manager # 特殊的字典 import multiprocessing # from multiprocessing import queues # 特殊隊列，可以讓進程間數據共享；只能在linux運行 import threading import time def foo(i,arg): arg[i] = i+100 for item in arg: print(item) print('=======================') if __name__ == "__main__": li = Array('i',10) # 這種方法不經常使用 for i in range(10): p = Process(target=foo,args=(i,li)) p.start() 輸出結果： 0 0 0 103 0 0 0 0 0 0 == == == == == == == == == == == = 0 0 102 103 0 0 0 0 0 0 == == == == == == == == == == == = 100 0 102 103 0 0 0 0 0 0 == == == == == == == == == == == = 100 0 102 103 104 0 0 0 0 0 == == == == == == == == == == == = 100 101 102 103 104 0 0 0 0 0 == == == == == == == == == == == = 100 101 102 103 104 0 106 0 0 0 == == == == == == == == == == == = 100 101 102 103 104 0 106 107 0 0 == == == == == == == == == == == = 100 101 102 103 104 0 106 107 108 0 == == == == == == == == == == == = 100 101 102 103 104 105 106 107 108 0 == == == == == == == == == == == = 100 101 102 103 104 105 106 107 108 109 == == == == == == == == == == == =
from multiprocessing import Process from multiprocessing import Array from multiprocessing import Manager # 特殊的字典 import multiprocessing # from multiprocessing import queues # 特殊隊列，可以讓進程間數據共享；只能在linux運行 import threading import time def foo(i, arg): print('say hi',i) arg[i] = i + 100 print(arg.values()) if __name__ == "__main__": # li = Array('i', 10) # 這種方法不經常使用 obj = Manager() li = obj.dict() # 經常使用方法 for i in range(10): p = Process(target=foo, args=(i, li)) # p.daemon = True p.start() p.join() print('end') 輸出結果： say hi 0 [100] say hi 1 [100, 101] say hi 2 [100, 101, 102] say hi 3 [100, 101, 102, 103] say hi 4 [100, 101, 102, 103, 104] say hi 5 [100, 101, 102, 103, 104, 105] say hi 6 [100, 101, 102, 103, 104, 105, 106] say hi 7 [100, 101, 102, 103, 104, 105, 106, 107] say hi 8 [100, 101, 102, 103, 104, 105, 106, 107, 108] say hi 9 [100, 101, 102, 103, 104, 105, 106, 107, 108, 109] end
11. 進程鎖
# 進程鎖，全部的進程同時去執行同一任務 from multiprocessing import Process,queues from multiprocessing import Array from multiprocessing import RLock,Lock,Condition,Semaphore import multiprocessing import time def foo(i,lis): lis[0] = lis[0] - 1 time.sleep(1) print('say hi',lis[0]) if __name__ == "__main__": li = Array('i',1) li[0] = 10 for i in range(10): p = Process(target=foo,args=(i,li)) # 建立10個進程，每一個進程都去執行foo方法 p.start() 輸出結果： say hi 0 say hi 0 say hi 0 say hi 0 say hi 0 say hi 0 say hi 0 say hi 0 say hi 0 say hi 0 # ########################################################################################### # 建立進程鎖，全部的進程逐個去執行同一任務 from multiprocessing import Process,queues from multiprocessing import Array from multiprocessing import RLock,Lock,Condition,Semaphore import multiprocessing import time def foo(i,lis,lc): # lc參數是接收下面線程中lock的參數 lc.acquire() # 加鎖 lis[0] = lis[0] - 1 time.sleep(1) print('say hi',lis[0]) lc.release() # 釋放鎖 if __name__ == "__main__": li = Array('i',1) li[0] = 10 lock = RLock() # 建立進程鎖，並傳遞給下面的lock參數 for i in range(10): p = Process(target=foo,args=(i,li,lock)) # 建立10個進程，每一個進程都去執行foo方法 p.start() 輸出結果： say hi 9 say hi 8 say hi 7 say hi 6 say hi 5 say hi 4 say hi 3 say hi 2 say hi 1 say hi 0
12. 線程池
進程池 from multiprocessing import Pool # 導入進程池 import time def f1(arg): time.sleep(1) print(arg) if __name__ == "__main__": pool = Pool(5) # 建立只有5個進程的池 for i in range(10): # 30個任務 pool.apply(func=f1,args=(i,)) # 每一個進程去執行f1 輸出結果： 0 1 2 3 4 5 6 7 8 9
# 進程池 from multiprocessing import Pool # 導入進程池 import time def f1(arg): time.sleep(1) print(arg) if __name__ == "__main__": pool = Pool(5) # 建立只有5個進程的池 for i in range(10): # 30個任務 pool.apply(func=f1,args=(i,)) # 每一個進程去執行f1 pool.apply_async(func=f1,args=(i,)) # apply表示去進程池取進程並執行串行操做 print('end') pool.close() # 表示全部的30個任務執行完畢才執行下面任務 time.sleep(1) # pool.terminet() # 當即終止任務執行 pool.join() #進程池中進程執行完畢後再關閉，若是註釋，那麼程序直接關閉。 輸出結果： 0 0 1 1 2 3 2 4 3 4 5 6 5 7 6 8 7 8 9 end 9
13. 協程的介紹
# 協程 # 原理：利用一個線程，分解一個線程成爲多個「微線程」(程序級別在作，與操做系統無關) # greenlet #本質上實現協程功能 # gevent # 內部封裝了底層功能greenlet,gevent是高級功能；是協程(高性能)，是人爲建立，只限I/O請求 from greenlet import greenlet def test1(): print(12) gr2.switch() # 切換做用 print(34) gr2.switch() def test2(): print(56) gr1.switch() print(78) gr1 = greenlet(test1) gr2 = greenlet(test2) gr1.switch() 輸出結果： 12 56 34 78 協程切換操做
import gevent def foo(): print('Running in foo') gevent.sleep(0) # 切換做用 print('Explicit context switch to foo again') def bar(): print('Explicit context to bar') gevent.sleep(0) print('Implicit context switch back to bar') gevent.joinall([ # 表示要執行下面兩個協程 gevent.spawn(foo), gevent.spawn(bar), ]) 輸出結果： Running in foo Explicit context to bar Explicit context switch to foo again Implicit context switch back to bar
# 協程發送http請求實例：一個線程併發發送3個http請求 # from gevent import monkey;monkey.patch_all() # 特殊socket請求發送後返回狀態 import gevent # 協程 import requests def func(url): print('GET: %s' % url) resp = requests.get(url) data = resp.text print('%d bytes receivd from %s' % (len(data),url)) gevent.joinall([ gevent.spawn(func,'http://www.python.org/'), gevent.spawn(func,'http://www.yahoo.com/'), gevent.spawn(func,'http://www.cnblogs.com'), gevent.spawn(func,'http://www.baidu.com'), ])
14. Python操做memcached和redis模塊介紹
memcached支持類型：
K-->"字符串"
redis支持類型：
K-->"字符串" 字符串
K-->[11,22,33,22] 列表
K-->{"k" : xxx} 字典
K-->[11,22,33] 集合
K-->[(11,1),(12,2),(13,3)] 排序集合
支持持久化功能
python操做mencached和redis需2個條件：
1.服務端安裝所需軟件
2.本地安裝其對應模塊(本質經過ssh或端口)
15. Memcached以及redis操做介紹
一. Memcached
Memcached 是一個高性能的分佈式內存對象緩存系統，用於動態Web應用以減輕數據庫負載。它經過在內存中緩存數據和對象來減小讀取數據庫的次數，從而提升動態、數據庫驅動網站的速度。Memcached基於一個存儲鍵/值對的hashmap。其守護進程（daemon ）是用C寫的，可是客戶端能夠用任何語言來編寫，並經過memcached協議與守護進程通訊。
Memcached安裝和基本使用
Memcached安裝：
wget http://memcached.org/latest
tar -zxvf memcached-1.x.x.tar.gz
cd memcached-1.x.x
./configure && make && make test && sudo make install
PS：依賴libevent
yum install libevent-devel
apt-get install libevent-dev
啓動Memcached
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命令
存儲命令: set/add/replace/append/prepend/cas 獲取命令: get/gets 其餘命令: delete/stats..
Python操做Memcached
安裝API
python操做Memcached使用Python-memcached模塊 下載安裝：https://pypi.python.org/pypi/python-memcached
一、第一次操做
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.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" 放置在該服務器的內存中
代碼實現以下：
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操做異常
#!/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不存在，則異常
#!/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存在，則修改
#!/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中刪除指定的多個鍵值對
#!/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 獲取多一個鍵值對
#!/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的值，在該值 前面 插入內容
#!/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 ）
#!/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 便可，如：
#!/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（獲取了緩衝的指定值）， 如此一來有可能出現非正常數據，則不容許修改。
二. redis
redis是一個key-value存儲系統。和Memcached相似，它支持存儲的value類型相對更多，包括string(字符串)、list(鏈表)、set(集合)、zset(sorted set --有序集合)和hash（哈希類型）。這些數據類型都支持push/pop、add/remove及取交集並集和差集及更豐富的操做，並且這些操做都是原子性的。在此基礎上，redis支持各類不一樣方式的排序。與memcached同樣，爲了保證效率，數據都是緩存在內存中。區別的是redis會週期性的把更新的數據寫入磁盤或者把修改操做寫入追加的記錄文件，而且在此基礎上實現了master-slave(主從)同步。
1. 使用Redis有哪些好處？ (1) 速度快，由於數據存在內存中，相似於HashMap，HashMap的優點就是查找和操做的時間複雜度都是O(1) (2) 支持豐富數據類型，支持string，list，set，sorted set，hash (3) 支持事務，操做都是原子性，所謂的原子性就是對數據的更改要麼所有執行，要麼所有不執行 (4) 豐富的特性：可用於緩存，消息，按key設置過時時間，過時後將會自動刪除 2. redis相比memcached有哪些優點？ (1) memcached全部的值均是簡單的字符串，redis做爲其替代者，支持更爲豐富的數據類型 (2) redis的速度比memcached快不少 (3) redis能夠持久化其數據 3. redis常見性能問題和解決方案： (1) Master最好不要作任何持久化工做，如RDB內存快照和AOF日誌文件 (2) 若是數據比較重要，某個Slave開啓AOF備份數據，策略設置爲每秒同步一次 (3) 爲了主從複製的速度和鏈接的穩定性，Master和Slave最好在同一個局域網內 (4) 儘可能避免在壓力很大的主庫上增長從庫 (5) 主從複製不要用圖狀結構，用單向鏈表結構更爲穩定，即：Master <- Slave1 <- Slave2 <- Slave3... 這樣的結構方便解決單點故障問題，實現Slave對Master的替換。若是Master掛了，能夠馬上啓用Slave1作Master，其餘不變。 4. MySQL裏有2000w數據，redis中只存20w的數據，如何保證redis中的數據都是熱點數據 相關知識：redis 內存數據集大小上升到必定大小的時候，就會施行數據淘汰策略。redis 提供 6種數據淘汰策略： voltile-lru：從已設置過時時間的數據集（server.db[i].expires）中挑選最近最少使用的數據淘汰 volatile-ttl：從已設置過時時間的數據集（server.db[i].expires）中挑選將要過時的數據淘汰 volatile-random：從已設置過時時間的數據集（server.db[i].expires）中任意選擇數據淘汰 allkeys-lru：從數據集（server.db[i].dict）中挑選最近最少使用的數據淘汰 allkeys-random：從數據集（server.db[i].dict）中任意選擇數據淘汰 no-enviction（驅逐）：禁止驅逐數據 5. Memcache與Redis的區別都有哪些？ 1)、存儲方式 Memecache把數據所有存在內存之中，斷電後會掛掉，數據不能超過內存大小。 Redis有部份存在硬盤上，這樣能保證數據的持久性。 2)、數據支持類型 Memcache對數據類型支持相對簡單。 Redis有複雜的數據類型。 3），value大小 redis最大能夠達到1GB，而memcache只有1MB 6. Redis 常見的性能問題都有哪些？如何解決？ 1).Master寫內存快照，save命令調度rdbSave函數，會阻塞主線程的工做，當快照比較大時對性能影響是很是大的，會間斷性暫停服務，因此Master最好不要寫內存快照。 2).Master AOF持久化，若是不重寫AOF文件，這個持久化方式對性能的影響是最小的，可是AOF文件會不斷增大，AOF文件過大會影響Master重啓的恢復速度。Master最好不要作任何持久化工做，包括內存快照和AOF日誌文件，特別是不要啓用內存快照作持久化,若是數據比較關鍵，某個Slave開啓AOF備份數據，策略爲每秒同步一次。 3).Master調用BGREWRITEAOF重寫AOF文件，AOF在重寫的時候會佔大量的CPU和內存資源，致使服務load太高，出現短暫服務暫停現象。 4). Redis主從複製的性能問題，爲了主從複製的速度和鏈接的穩定性，Slave和Master最好在同一個局域網內 7, redis 最適合的場景 Redis最適合全部數據in-momory的場景，雖然Redis也提供持久化功能，但實際更多的是一個disk-backed的功能，跟傳統意義上的持久化有比較大的差異，那麼可能你們就會有疑問，彷佛Redis更像一個增強版的Memcached，那麼什麼時候使用Memcached,什麼時候使用Redis呢? 若是簡單地比較Redis與Memcached的區別，大多數都會獲得如下觀點： 、Redis不只僅支持簡單的k/v類型的數據，同時還提供list，set，zset，hash等數據結構的存儲。 、Redis支持數據的備份，即master-slave模式的數據備份。 、Redis支持數據的持久化，能夠將內存中的數據保持在磁盤中，重啓的時候能夠再次加載進行使用。 （1）、會話緩存（Session Cache） 最經常使用的一種使用Redis的情景是會話緩存（session cache）。用Redis緩存會話比其餘存儲（如Memcached）的優點在於：Redis提供持久化。當維護一個不是嚴格要求一致性的緩存時，若是用戶的購物車信息所有丟失，大部分人都會不高興的，如今，他們還會這樣嗎？ 幸運的是，隨着 Redis 這些年的改進，很容易找到怎麼恰當的使用Redis來緩存會話的文檔。甚至廣爲人知的商業平臺Magento也提供Redis的插件。 （2）、全頁緩存（FPC） 除基本的會話token以外，Redis還提供很簡便的FPC平臺。回到一致性問題，即便重啓了Redis實例，由於有磁盤的持久化，用戶也不會看到頁面加載速度的降低，這是一個極大改進，相似PHP本地FPC。 再次以Magento爲例，Magento提供一個插件來使用Redis做爲全頁緩存後端。 此外，對WordPress的用戶來講，Pantheon有一個很是好的插件 wp-redis，這個插件能幫助你以最快速度加載你曾瀏覽過的頁面。 （3）、隊列 Reids在內存存儲引擎領域的一大優勢是提供 list 和 set 操做，這使得Redis能做爲一個很好的消息隊列平臺來使用。Redis做爲隊列使用的操做，就相似於本地程序語言（如Python）對 list 的 push/pop 操做。 若是你快速的在Google中搜索「Redis queues」，你立刻就能找到大量的開源項目，這些項目的目的就是利用Redis建立很是好的後端工具，以知足各類隊列需求。例如，Celery有一個後臺就是使用Redis做爲broker，你能夠從這裏去查看。 （4），排行榜/計數器 Redis在內存中對數字進行遞增或遞減的操做實現的很是好。集合（Set）和有序集合（Sorted Set）也使得咱們在執行這些操做的時候變的很是簡單，Redis只是正好提供了這兩種數據結構。因此，咱們要從排序集合中獲取到排名最靠前的10個用戶–咱們稱之爲「user_scores」，咱們只須要像下面同樣執行便可： 固然，這是假定你是根據你用戶的分數作遞增的排序。若是你想返回用戶及用戶的分數，你須要這樣執行： ZRANGE user_scores 0 10 WITHSCORES Agora Games就是一個很好的例子，用Ruby實現的，它的排行榜就是使用Redis來存儲數據的，你能夠在這裏看到。 （5）、發佈/訂閱 最後（但確定不是最不重要的）是Redis的發佈/訂閱功能。發佈/訂閱的使用場景確實很是多。我已看見人們在社交網絡鏈接中使用，還可做爲基於發佈/訂閱的腳本觸發器，甚至用Redis的發佈/訂閱功能來創建聊天系統！（不，這是真的，你能夠去核實）。 Redis提供的全部特性中，我感受這個是喜歡的人最少的一個，雖然它爲用戶提供若是此多功能。
- Redis安裝和基本使用
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 redis> set foo bar OK redis> get foo "bar"
- Python操做Redis
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。
#!/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實例共享一個鏈接池。
#!/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) 在Redis中設置值，默認，不存在則建立，存在則修改
參數：
ex，過時時間（秒）
px，過時時間（毫秒）
nx，若是設置爲True，則只有name不存在時，當前set操做才執行
xx，若是設置爲True，則只有name存在時，崗前set操做才執行
setnx(name, value) 設置值，只有name不存在時，執行設置操做（添加） setex(name, value, time) # 設置值
# 參數：
# time，過時時間（數字秒 或 timedelta對象）
mset(*args, **kwargs) 批量設置值
如：
mset(k1='v1', k2='v2')
或
mget({'k1': 'v1', 'k2': 'v2'})
get(name) 獲取值 mget(keys, *args) 批量獲取
如：
mget('ylr', 'wupeiqi')
或
r.mget(['ylr', 'wupeiqi'])
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"
# 擴展，轉換二進制表示：
# 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) # 獲取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,自增數（必須是整數）
# 注：同incrby
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) 表示從右向左操做
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) # 同字符串的操做，用於增量迭代分批獲取元素，避免內存消耗太大 |
有序集合，在集合的基礎上，爲每元素排序；元素的排序須要根據另一個值來進行比較，因此，對於有序集合，每個元素有兩個值，即：值和分數，分數專門用來作排序。
| 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) # 同字符串類似，相較於字符串新增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
# 爲某個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) # 同字符串操做，用於增量迭代獲取key
|
四、管道
redis-py默認在執行每次請求都會建立（鏈接池申請鏈接）和斷開（歸還鏈接池）一次鏈接操做，若是想要在一次請求中指定多個命令，則可使用pipline實現一次請求指定多個命令，而且默認狀況下一次pipline 是原子性操做。
#!/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.multi()
pipe.set('name', 'alex')
pipe.set('role', 'sb')
pipe.execute()
#!/usr/bin/env python # -*- coding:utf-8 -*- import redis conn = redis.Redis(host='192.168.1.41',port=6379) conn.set('count',1000) with conn.pipeline() as pipe: # 先監視，本身的值沒有被修改過 conn.watch('count') # 事務開始 pipe.multi() old_count = conn.get('count') count = int(old_count) if count > 0: # 有庫存 pipe.set('count', count - 1) # 執行，把全部命令一次性推送過去 pipe.execute()
五、發佈訂閱
發佈者：服務器
訂閱者：Dashboad和數據處理
Demo以下：
#!/usr/bin/env python # -*- coding:utf-8 -*- import redis class RedisHelper: def __init__(self): self.__conn = redis.Redis(host='10.211.55.4') self.chan_sub = 'fm104.5' self.chan_pub = 'fm104.5' def public(self, msg): self.__conn.publish(self.chan_pub, msg) return True def subscribe(self): pub = self.__conn.pubsub() pub.subscribe(self.chan_sub) pub.parse_response() return pub
訂閱者：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from monitor.RedisHelper import RedisHelper
obj = RedisHelper()
redis_sub = obj.subscribe()
while True:
msg= redis_sub.parse_response()
print msg
發佈者：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from monitor.RedisHelper import RedisHelper
obj = RedisHelper()
obj.public('hello')
6. sentinel
redis重的sentinel主要用於在redis主從複製中，若是master顧上，則自動將slave替換成master
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from redis.sentinel import Sentinel
# 鏈接哨兵服務器(主機名也能夠用域名)
sentinel = Sentinel([('10.211.55.20', 26379),
('10.211.55.20', 26380),
],
socket_timeout=0.5)
# # 獲取主服務器地址
# master = sentinel.discover_master('mymaster')
# print(master)
#
# # # 獲取從服務器地址
# slave = sentinel.discover_slaves('mymaster')
# print(slave)
#
#
# # # 獲取主服務器進行寫入
# master = sentinel.master_for('mymaster')
# master.set('foo', 'bar')
# # # # 獲取從服務器進行讀取（默認是round-roubin）
# slave = sentinel.slave_for('mymaster', password='redis_auth_pass')
# r_ret = slave.get('foo')
# print(r_ret)
更多參見：https://github.com/andymccurdy/redis-py/
http://doc.redisfans.com/
16. 線程池重點講解
17. 上下文管理
18. redis知識拾遺之發佈訂閱
第十章.rabbitMQ、MySQL 、SQLAlchemy和Paramiko的基本使用
1. 初識rabbitMQ
RabbitMQ是一個在AMQP基礎上完整的，可複用的企業消息系統。他遵循Mozilla Public License開源協議。
MQ全稱爲Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通訊方法。應用程序經過讀寫出入隊列的消息（針對應用程序的數據）來通訊，而無需專用鏈接來連接它們。消 息傳遞指的是程序之間經過在消息中發送數據進行通訊，而不是經過直接調用彼此來通訊，直接調用一般是用於諸如遠程過程調用的技術。排隊指的是應用程序經過 隊列來通訊。隊列的使用除去了接收和發送應用程序同時執行的要求。
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
安裝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()
2. rabbitMQ的使用
對於RabbitMQ來講，生產和消費再也不針對內存裏的一個Queue對象，而是某臺服務器上的RabbitMQ Server實現的消息隊列。
#!/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()
# ########################## 消費者 ##########################
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會從新將該任務添加到隊列中。
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 消息不丟失
#!/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 去隊列中獲取 奇數 序列的任務，消費者1去隊列中獲取 偶數 序列的任務。
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() 訂閱者
五、關鍵字發送
exchange type = direct
以前事例，發送消息時明確指定某個隊列並向其中發送消息，RabbitMQ還支持根據關鍵字發送，即：隊列綁定關鍵字，發送者將數據根據關鍵字發送到消息exchange，exchange根據 關鍵字 斷定應該將數據發送至指定隊列。
#!/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 個 或 多個 單詞
- * 表示只能匹配 一個 單詞
發送者路由值 隊列中 old.boy.python old.* -- 不匹配 old.boy.python old.# -- 匹配
#!/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() 生產者
注意：
sudo rabbitmqctl add_user alex 123 # 設置用戶爲administrator角色 sudo rabbitmqctl set_user_tags alex administrator # 設置權限 sudo rabbitmqctl set_permissions -p "/" alex '.''.''.' # 而後重啓rabbiMQ服務 sudo /etc/init.d/rabbitmq-server restart # 而後可使用剛纔的用戶遠程鏈接rabbitmq server了。 ------------------------------ credentials = pika.PlainCredentials("alex","123") connection = pika.BlockingConnection(pika.ConnectionParameters('192.168.14.47',credentials=credentials))
3. MySQL介紹
MySQL是一個關係型數據庫管理系統，由瑞典MySQL AB 公司開發，目前屬於 Oracle 旗下公司。MySQL 最流行的關係型數據庫管理系統，在 WEB 應用方面MySQL是最好的 RDBMS (Relational Database Management System，關係數據庫管理系統) 應用軟件之一。
想要使用MySQL來存儲並操做數據，則須要作幾件事情：
a. 安裝MySQL服務端
b. 安裝MySQL客戶端
b. 【客戶端】鏈接【服務端】
c. 【客戶端】發送命令給【服務端MySQL】服務的接受命令並執行相應操做(增刪改查等)
下載
http://dev.mysql.com/downloads/mysql/
安裝
windows:
點點點
Linux:
yum install mysql-server
Mac:
點點點
Window版本
一、下載
MySQL Community Server 5.7.16 http://dev.mysql.com/downloads/mysql/
二、解壓
若是想要讓MySQL安裝在指定目錄，那麼就將解壓後的文件夾移動到指定目錄，如：C:\mysql-5.7.16-winx64
三、初始化
MySQL解壓後的 bin 目錄下有一大堆的可執行文件，執行以下命令初始化數據：
cd c:\mysql-5.7.16-winx64\bin mysqld --initialize-insecure
四、啓動MySQL服務
執行命令從而啓動MySQL服務
# 進入可執行文件目錄 cd c:\mysql-5.7.16-winx64\bin # 啓動MySQL服務 mysqld
五、啓動MySQL客戶端並鏈接MySQL服務
因爲初始化時使用的【mysqld --initialize-insecure】命令，其默認未給root帳戶設置密碼
# 進入可執行文件目錄 cd c:\mysql-5.7.16-winx64\bin # 鏈接MySQL服務器 mysql -u root -p # 提示請輸入密碼，直接回車
輸入回車，見下圖表示安裝成功：
到此爲止，MySQL服務端已經安裝成功而且客戶端已經能夠鏈接上，之後再操做MySQL時，只須要重複上述四、5步驟便可。可是，在四、5步驟中重複的進入可執行文件目錄比較繁瑣，如想往後操做簡便，能夠作以下操做。
| a. 添加環境變量 將MySQL可執行文件添加到環境變量中，從而執行執行命令便可 【右鍵計算機】--》【屬性】--》【高級系統設置】--》【高級】--》【環境變量】--》【在第二個內容框中找到 變量名爲Path 的一行，雙擊】 --> 【將MySQL的bin目錄路徑追加到變值值中，用 ； 分割】 如： C:\Program Files (x86)\Parallels\Parallels Tools\Applications;%SystemRoot%\system32;%SystemRoot%;%SystemRoot%\System32\Wbem;%SYSTEMROOT%\System32\WindowsPowerShell\v1.0\;C:\Python27;C:\Python35;C:\mysql-5.7.16-winx64\bin 如此一來，之後再啓動服務並鏈接時，僅需： # 啓動MySQL服務，在終端輸入 mysqld # 鏈接MySQL服務，在終端輸入： mysql -u root -p b. 將MySQL服務製做成windows服務 上一步解決了一些問題，但不夠完全，由於在執行【mysqd】啓動MySQL服務器時，當前終端會被hang住，那麼作一下設置便可解決此問題： # 製做MySQL的Windows服務，在終端執行此命令： "c:\mysql-5.7.16-winx64\bin\mysqld" --install # 移除MySQL的Windows服務，在終端執行此命令： "c:\mysql-5.7.16-winx64\bin\mysqld" --remove 註冊成服務以後，之後再啓動和關閉MySQL服務時，僅需執行以下命令： # 啓動MySQL服務 net start mysql # 關閉MySQL服務 net stop mysql
|
Linux版本
安裝：
yum install mysql-server
服務端啓動
mysql.server start
客戶端鏈接
鏈接：
mysql -h host -u user -p
常見錯誤：
ERROR 2002 (HY000): Can't connect to local MySQL server through socket '/tmp/mysql.sock' (2), it means that the MySQL server daemon (Unix) or service (Windows) is not running.
退出：
QUIT 或者 Control+D
4. MySQL數據庫及表操做
一、顯示數據庫
SHOW DATABASES;
默認數據庫：
mysql - 用戶權限相關數據
test - 用於用戶測試數據
information_schema - MySQL自己架構相關數據
二、建立數據庫
# utf-8 CREATE DATABASE 數據庫名稱 DEFAULT CHARSET utf8 COLLATE utf8_general_ci; # gbk CREATE DATABASE 數據庫名稱 DEFAULT CHARACTER SET gbk COLLATE gbk_chinese_ci;
三、使用數據庫
USE db_name;
顯示當前使用的數據庫中全部表：SHOW TABLES;
四、用戶管理
建立用戶
create user '用戶名'@'IP地址' identified by '密碼';
刪除用戶
drop user '用戶名'@'IP地址';
修改用戶
rename user '用戶名'@'IP地址'; to '新用戶名'@'IP地址';;
修改密碼
set password for '用戶名'@'IP地址' = Password('新密碼')
PS：用戶權限相關數據保存在mysql數據庫的user表中，因此也能夠直接對其進行操做（不建議）
五、受權管理
show grants for '用戶'@'IP地址' -- 查看權限 grant 權限 on 數據庫.表 to '用戶'@'IP地址' -- 受權 revoke 權限 on 數據庫.表 from '用戶'@'IP地址' -- 取消權限
all privileges 除grant外的全部權限 select 僅查權限 select,insert 查和插入權限 ... usage 無訪問權限 alter 使用alter table alter routine 使用alter procedure和drop procedure create 使用create table create routine 使用create procedure create temporary tables 使用create temporary tables create user 使用create user、drop user、rename user和revoke all privileges create view 使用create view delete 使用delete drop 使用drop table execute 使用call和存儲過程 file 使用select into outfile 和 load data infile grant option 使用grant 和 revoke index 使用index insert 使用insert lock tables 使用lock table process 使用show full processlist select 使用select show databases 使用show databases show view 使用show view update 使用update reload 使用flush shutdown 使用mysqladmin shutdown(關閉MySQL) super 使用change master、kill、logs、purge、master和set global。還容許mysqladmin調試登錄 replication client 服務器位置的訪問 replication slave 由複製從屬使用 對於權限
對於目標數據庫以及內部其餘： 數據庫名.* 數據庫中的全部 數據庫名.表 指定數據庫中的某張表 數據庫名.存儲過程 指定數據庫中的存儲過程 *.* 全部數據庫
用戶名@IP地址 用戶只能在改IP下才能訪問 用戶名@192.168.1.% 用戶只能在改IP段下才能訪問(通配符%表示任意) 用戶名@% 用戶能夠再任意IP下訪問(默認IP地址爲%)
grant all privileges on db1.tb1 TO '用戶名'@'IP' grant select on db1.* TO '用戶名'@'IP' grant select,insert on *.* TO '用戶名'@'IP' revoke select on db1.tb1 from '用戶名'@'IP'
特殊的：
flush privileges，將數據讀取到內存中，從而當即生效。
# 啓動免受權服務端 mysqld --skip-grant-tables # 客戶端 mysql -u root -p # 修改用戶名密碼 update mysql.user set authentication_string=password('666') where user='root'; flush privileges;
5. MySQL數據的CURD
一、建立表
create table 表名(
列名 類型 是否能夠爲空，
列名 類型 是否能夠爲空
)ENGINE=InnoDB DEFAULT CHARSET=utf8
是否可空，null表示空，非字符串 not null - 不可空 null - 可空
默認值，建立列時能夠指定默認值，當插入數據時若是未主動設置，則自動添加默認值 create table tb1( nid int not null defalut 2, num int not null )
自增，若是爲某列設置自增列，插入數據時無需設置此列，默認將自增（表中只能有一個自增列） create table tb1( nid int not null auto_increment primary key, num int null ) 或 create table tb1( nid int not null auto_increment, num int null, index(nid) ) 注意：1、對於自增列，必須是索引（含主鍵）。 2、對於自增能夠設置步長和起始值 show session variables like 'auto_inc%'; set session auto_increment_increment=2; set session auto_increment_offset=10; shwo global variables like 'auto_inc%'; set global auto_increment_increment=2; set global auto_increment_offset=10; 自增
主鍵，一種特殊的惟一索引，不容許有空值，若是主鍵使用單個列，則它的值必須惟一，若是是多列，則其組合必須惟一。 create table tb1( nid int not null auto_increment primary key, num int null ) 或 create table tb1( nid int not null, num int not null, primary key(nid,num) ) 主鍵
外鍵，一個特殊的索引，只能是指定內容 creat table color( nid int not null primary key, name char(16) not null ) create table fruit( nid int not null primary key, smt char(32) null , color_id int not null, constraint fk_cc foreign key (color_id) references color(nid) ) 外鍵
二、刪除表
drop table 表名
三、清空表
delete from 表名 truncate table 表名
四、修改表
添加列：alter table 表名 add 列名 類型
刪除列：alter table 表名 drop column 列名
修改列：
alter table 表名 modify column 列名 類型; -- 類型
alter table 表名 change 原列名 新列名 類型; -- 列名，類型
添加主鍵：
alter table 表名 add primary key(列名);
刪除主鍵：
alter table 表名 drop primary key;
alter table 表名 modify 列名 int, drop primary key;
添加外鍵：alter table 從表 add constraint 外鍵名稱（形如：FK_從表_主表） foreign key 從表(外鍵字段) references 主表(主鍵字段);
刪除外鍵：alter table 表名 drop foreign key 外鍵名稱
修改默認值：ALTER TABLE testalter_tbl ALTER i SET DEFAULT 1000;
刪除默認值：ALTER TABLE testalter_tbl ALTER i DROP DEFAULT;
五、基本數據類型
MySQL的數據類型大體分爲：數值、時間和字符串
bit[(M)] 二進制位（101001），m表示二進制位的長度（1-64），默認m＝1 tinyint[(m)] [unsigned] [zerofill] 小整數，數據類型用於保存一些範圍的整數數值範圍： 有符號： -128 ～ 127. 無符號： ～ 255 特別的： MySQL中無布爾值，使用tinyint(1)構造。 int[(m)][unsigned][zerofill] 整數，數據類型用於保存一些範圍的整數數值範圍： 有符號： -2147483648 ～ 2147483647 無符號： ～ 4294967295 特別的：整數類型中的m僅用於顯示，對存儲範圍無限制。例如： int(5),當插入數據2時，select 時數據顯示爲： 00002 bigint[(m)][unsigned][zerofill] 大整數，數據類型用於保存一些範圍的整數數值範圍： 有符號： -9223372036854775808 ～ 9223372036854775807 無符號： ～ 18446744073709551615 decimal[(m[,d])] [unsigned] [zerofill] 準確的小數值，m是數字總個數（負號不算），d是小數點後個數。 m最大值爲65，d最大值爲30。 特別的：對於精確數值計算時須要用此類型 decaimal可以存儲精確值的緣由在於其內部按照字符串存儲。 FLOAT[(M,D)] [UNSIGNED] [ZEROFILL] 單精度浮點數（非準確小數值），m是數字總個數，d是小數點後個數。 無符號： -3.402823466E+38 to -1.175494351E-38, 1.175494351E-38 to 3.402823466E+38 有符號： 1.175494351E-38 to 3.402823466E+38 **** 數值越大，越不許確 **** DOUBLE[(M,D)] [UNSIGNED] [ZEROFILL] 雙精度浮點數（非準確小數值），m是數字總個數，d是小數點後個數。 無符號： -1.7976931348623157E+308 to -2.2250738585072014E-308 2.2250738585072014E-308 to 1.7976931348623157E+308 有符號： 2.2250738585072014E-308 to 1.7976931348623157E+308 **** 數值越大，越不許確 **** char (m) char數據類型用於表示固定長度的字符串，能夠包含最多達255個字符。其中m表明字符串的長度。 PS: 即便數據小於m長度，也會佔用m長度 varchar(m) varchars數據類型用於變長的字符串，能夠包含最多達255個字符。其中m表明該數據類型所容許保存的字符串的最大長度，只要長度小於該最大值的字符串均可以被保存在該數據類型中。 注：雖然varchar使用起來較爲靈活，可是從整個系統的性能角度來講，char數據類型的處理速度更快，有時甚至能夠超出varchar處理速度的50%。所以，用戶在設計數據庫時應當綜合考慮各方面的因素，以求達到最佳的平衡 text text數據類型用於保存變長的大字符串，能夠組多到65535 (2**16 − 1)個字符。 mediumtext A TEXT column with a maximum length of 16,777,215 (2**24 − 1) characters. longtext A TEXT column with a maximum length of 4,294,967,295 or 4GB (2**32 − 1) characters. enum 枚舉類型， An ENUM column can have a maximum of 65,535 distinct elements. (The practical limit is less than 3000.) 示例： CREATE TABLE shirts ( name VARCHAR(40), size ENUM('x-small', 'small', 'medium', 'large', 'x-large') ); INSERT INTO shirts (name, size) VALUES ('dress shirt','large'), ('t-shirt','medium'),('polo shirt','small'); set 集合類型 A SET column can have a maximum of 64 distinct members. 示例： CREATE TABLE myset (col SET('a', 'b', 'c', 'd')); INSERT INTO myset (col) VALUES ('a,d'), ('d,a'), ('a,d,a'), ('a,d,d'), ('d,a,d'); DATE YYYY-MM-DD（1000-01-01/9999-12-31） TIME HH:MM:SS（'-838:59:59'/'838:59:59'） YEAR YYYY（1901/2155） DATETIME YYYY-MM-DD HH:MM:SS（1000-01-01 00:00:00/9999-12-31 23:59:59 Y） TIMESTAMP YYYYMMDD HHMMSS（1970-01-01 00:00:00/2037 年某時）
二進制數據：TinyBlob、Blob、MediumBlob、LongBlob
更多參考：
- http://www.runoob.com/mysql/mysql-data-types.html
- http://dev.mysql.com/doc/refman/5.7/en/data-type-overview.html
六、表內容操做
- 增
insert into 表 (列名,列名...) values (值,值,值...) insert into 表 (列名,列名...) values (值,值,值...),(值,值,值...) insert into 表 (列名,列名...) select (列名,列名...) from 表
- 刪
delete from 表 delete from 表 where id＝1 and name＝'alex'
- 改
update 表 set name ＝ 'alex' where id>1
- 查
select * from 表 select * from 表 where id > 1 select nid,name,gender as gg from 表 where id > 1
- 其餘
a、條件
select * from 表 where id > 1 and name != 'alex' and num = 12;
select * from 表 where id between 5 and 16;
select * from 表 where id in (11,22,33)
select * from 表 where id not in (11,22,33)
select * from 表 where id in (select nid from 表)
b、通配符
select * from 表 where name like 'ale%' - ale開頭的全部（多個字符串）
select * from 表 where name like 'ale_' - ale開頭的全部（一個字符）
c、限制
select * from 表 limit 5; - 前5行
select * from 表 limit 4,5; - 從第4行開始的5行
select * from 表 limit 5 offset 4 - 從第4行開始的5行
d、排序
select * from 表 order by 列 asc - 根據 「列」 從小到大排列
select * from 表 order by 列 desc - 根據 「列」 從大到小排列
select * from 表 order by 列1 desc,列2 asc - 根據 「列1」 從大到小排列，若是相同則按列2從小到大排序
e、分組
select num from 表 group by num
select num,nid from 表 group by num,nid
select num,nid from 表 where nid > 10 group by num,nid order nid desc
select num,nid,count(*),sum(score),max(score),min(score) from 表 group by num,nid
select num from 表 group by num having max(id) > 10
特別的：group by 必須在where以後，order by以前
f、連表
無對應關係則不顯示
select A.num, A.name, B.name
from A,B
Where A.nid = B.nid
無對應關係則不顯示
select A.num, A.name, B.name
from A inner join B
on A.nid = B.nid
A表全部顯示，若是B中無對應關係，則值爲null
select A.num, A.name, B.name
from A left join B
on A.nid = B.nid
B表全部顯示，若是B中無對應關係，則值爲null
select A.num, A.name, B.name
from A right join B
on A.nid = B.nid
g、組合
組合，自動處理重合
select nickname
from A
union
select name
from B
組合，不處理重合
select nickname
from A
union all
select name
from B
6. pymsql模塊操做mysql
Python操做MySQL主要使用兩種方式：
- 原生模塊 pymsql
- ORM框架 SQLAchemy
pymsql是Python中操做MySQL的模塊，其使用方法和MySQLdb幾乎相同。
下載安裝
pip3 install pymysql
使用操做
一、執行SQL
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
# 建立鏈接
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
# 建立遊標
cursor = conn.cursor()
# 執行SQL，並返回收影響行數
effect_row = cursor.execute("update hosts set host = '1.1.1.2'")
# 執行SQL，並返回受影響行數
#effect_row = cursor.execute("update hosts set host = '1.1.1.2' where nid > %s", (1,))
# 執行SQL，並返回受影響行數
#effect_row = cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
# 提交，否則沒法保存新建或者修改的數據
conn.commit()
# 關閉遊標
cursor.close()
# 關閉鏈接
conn.close()
二、獲取新建立數據自增ID
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor = conn.cursor()
cursor.executemany("insert into hosts(host,color_id)values(%s,%s)", [("1.1.1.11",1),("1.1.1.11",2)])
conn.commit()
cursor.close()
conn.close()
# 獲取最新自增ID
new_id = cursor.lastrowid
三、獲取查詢數據
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
cursor = conn.cursor()
cursor.execute("select * from hosts")
# 獲取第一行數據
row_1 = cursor.fetchone()
# 獲取前n行數據
# row_2 = cursor.fetchmany(3)
# 獲取全部數據
# row_3 = cursor.fetchall()
conn.commit()
cursor.close()
conn.close()
注：在fetch數據時按照順序進行，可使用cursor.scroll(num,mode)來移動遊標位置，如：
- cursor.scroll(1,mode='relative') # 相對當前位置移動
- cursor.scroll(2,mode='absolute') # 相對絕對位置移動
四、fetch數據類型
關於默認獲取的數據是元祖類型，若是想要或者字典類型的數據，即：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import pymysql
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='123', db='t1')
# 遊標設置爲字典類型
cursor = conn.cursor(cursor=pymysql.cursors.DictCursor)
r = cursor.execute("call p1()")
result = cursor.fetchone()
conn.commit()
cursor.close()
conn.close()
加一個示例：
7. SQLAlchemy介紹
SQLAlchemy是Python編程語言下的一款ORM框架，該框架創建在數據庫API之上，使用關係對象映射進行數據庫操做，簡言之即是：將對象轉換成SQL，而後使用數據API執行SQL並獲取執行結果。
Dialect用於和數據API進行交流，根據配置文件的不一樣調用不一樣的數據庫API，從而實現對數據庫的操做，如：
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
8. SQLAlchemy之建立表
9. SQLAlchemy之CURD
步驟一：
使用 Engine/ConnectionPooling/Dialect 進行數據庫操做，Engine使用ConnectionPooling鏈接數據庫，而後再經過Dialect執行SQL語句。
#!/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_engine engine = 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，並獲取結果。
#!/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, 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) 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。
#!/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()
10. SQLAlchemy之一對多關係
11. SQLAlchemy表建立過程
12. SQLAlchemy之多對多關係
13. SQLAlchemy內容梳理以及學習流程
14. Paramiko模塊介紹以及自定義封裝
Python的paramiko模塊，該模塊基於SSH用於鏈接遠程服務器並執行相關操做
SSHClient
用於鏈接遠程服務器並執行基本命令
基於用戶名密碼鏈接：
import paramiko
# 建立SSH對象
ssh = paramiko.SSHClient()
# 容許鏈接不在know_hosts文件中的主機
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 鏈接服務器
ssh.connect(hostname='c1.salt.com', port=22, username='wupeiqi', password='123')
# 執行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 獲取命令結果
result = stdout.read()
# 關閉鏈接
ssh.close()
import paramiko transport = paramiko.Transport(('hostname', 22)) transport.connect(username='wupeiqi', password='123') ssh = paramiko.SSHClient() ssh._transport = transport stdin, stdout, stderr = ssh.exec_command('df') print stdout.read() transport.close()
基於公鑰密鑰鏈接：
import paramiko
private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa')
# 建立SSH對象
ssh = paramiko.SSHClient()
# 容許鏈接不在know_hosts文件中的主機
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 鏈接服務器
ssh.connect(hostname='c1.salt.com', port=22, username='wupeiqi', key=private_key)
# 執行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 獲取命令結果
result = stdout.read()
# 關閉鏈接
ssh.close()
import paramiko private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa') transport = paramiko.Transport(('hostname', 22)) transport.connect(username='wupeiqi', pkey=private_key) ssh = paramiko.SSHClient() ssh._transport = transport stdin, stdout, stderr = ssh.exec_command('df') transport.close()
import paramiko from io import StringIO key_str = """-----BEGIN RSA PRIVATE KEY----- MIIEpQIBAAKCAQEAq7gLsqYArAFco02/55IgNg0r7NXOtEM3qXpb/dabJ5Uyky/8 NEHhFiQ7deHIRIuTW5Zb0kD6h6EBbVlUMBmwJrC2oSzySLU1w+ZNfH0PE6W6fans H80whhuc/YgP+fjiO+VR/gFcqib8Rll5UfYzf5H8uuOnDeIXGCVgyHQSmt8if1+e 7hn1MVO1Lrm9Fco8ABI7dyv8/ZEwoSfh2C9rGYgA58LT1FkBRkOePbHD43xNfAYC tfLvz6LErMnwdOW4sNMEWWAWv1fsTB35PAm5CazfKzmam9n5IQXhmUNcNvmaZtvP c4f4g59mdsaWNtNaY96UjOfx83Om86gmdkKcnwIDAQABAoIBAQCnDBGFJuv8aA7A ZkBLe+GN815JtOyye7lIS1n2I7En3oImoUWNaJEYwwJ8+LmjxMwDCtAkR0XwbvY+ c+nsKPEtkjb3sAu6I148RmwWsGncSRqUaJrljOypaW9dS+GO4Ujjz3/lw1lrxSUh IqVc0E7kyRW8kP3QCaNBwArYteHreZFFp6XmtKMtXaEA3saJYILxaaXlYkoRi4k8 S2/K8aw3ZMR4tDCOfB4o47JaeiA/e185RK3A+mLn9xTDhTdZqTQpv17/YRPcgmwz zu30fhVXQT/SuI0sO+bzCO4YGoEwoBX718AWhdLJFoFq1B7k2ZEzXTAtjEXQEWm6 01ndU/jhAasdfasdasdfasdfa3eraszxqwefasdfadasdffsFIfAsjQb4HdkmHuC OeJrJOd+CYvdEeqJJNnF6AbHyYHIECkj0Qq1kEfLOEsqzd5nDbtkKBte6M1trbjl HtJ2Yb8w6o/q/6Sbj7wf/cW3LIYEdeVCjScozVcQ9R83ea05J+QOAr4nAoGBAMaq UzLJfLNWZ5Qosmir2oHStFlZpxspax/ln7DlWLW4wPB4YJalSVovF2Buo8hr8X65 lnPiE41M+G0Z7icEXiFyDBFDCtzx0x/RmaBokLathrFtI81UCx4gQPLaSVNMlvQA 539GsubSrO4LpHRNGg/weZ6EqQOXvHvkUkm2bDDJAoGATytFNxen6GtC0ZT3SRQM WYfasdf3xbtuykmnluiofasd2sfmjnljkt7khghmghdasSDFGQfgaFoKfaawoYeH C2XasVUsVviBn8kPSLSVBPX4JUfQmA6h8HsajeVahxN1U9e0nYJ0sYDQFUMTS2t8 RT57+WK/0ONwTWHdu+KnaJECgYEAid/ta8LQC3p82iNAZkpWlGDSD2yb/8rH8NQg 9tjEryFwrbMtfX9qn+8srx06B796U3OjifstjJQNmVI0qNlsJpQK8fPwVxRxbJS/ pMbNICrf3sUa4sZgDOFfkeuSlgACh4cVIozDXlR59Z8Y3CoiW0uObEgvMDIfenAj 98pl3ZkCgYEAj/UCSni0dwX4pnKNPm6LUgiS7QvIgM3H9piyt8aipQuzBi5LUKWw DlQC4Zb73nHgdREtQYYXTu7p27Bl0Gizz1sW2eSgxFU8eTh+ucfVwOXKAXKU5SeI +MbuBfUYQ4if2N/BXn47+/ecf3A4KgB37Le5SbLDddwCNxGlBzbpBa0= -----END RSA PRIVATE KEY-----""" private_key = paramiko.RSAKey(file_obj=StringIO(key_str)) transport = paramiko.Transport(('10.0.1.40', 22)) transport.connect(username='wupeiqi', pkey=private_key) ssh = paramiko.SSHClient() ssh._transport = transport stdin, stdout, stderr = ssh.exec_command('df') result = stdout.read() transport.close() print(result) 基於私鑰字符串進行鏈接
SFTPClient
用於鏈接遠程服務器並執行上傳下載
基於用戶名密碼上傳下載
import paramiko
transport = paramiko.Transport(('hostname',22))
transport.connect(username='wupeiqi',password='123')
sftp = paramiko.SFTPClient.from_transport(transport)
# 將location.py 上傳至服務器 /tmp/test.py
sftp.put('/tmp/location.py', '/tmp/test.py')
# 將remove_path 下載到本地 local_path
sftp.get('remove_path', 'local_path')
transport.close()
基於公鑰密鑰上傳下載
import paramiko
private_key = paramiko.RSAKey.from_private_key_file('/home/auto/.ssh/id_rsa')
transport = paramiko.Transport(('hostname', 22))
transport.connect(username='wupeiqi', pkey=private_key )
sftp = paramiko.SFTPClient.from_transport(transport)
# 將location.py 上傳至服務器 /tmp/test.py
sftp.put('/tmp/location.py', '/tmp/test.py')
# 將remove_path 下載到本地 local_path
sftp.get('remove_path', 'local_path')
transport.close()
#!/usr/bin/env python # -*- coding:utf-8 -*- import paramiko import uuid class Haproxy(object): def __init__(self): self.host = '172.16.103.191' self.port = 22 self.username = 'wupeiqi' self.pwd = '123' self.__k = None def create_file(self): file_name = str(uuid.uuid4()) with open(file_name,'w') as f: f.write('sb') return file_name def run(self): self.connect() self.upload() self.rename() self.close() def connect(self): transport = paramiko.Transport((self.host,self.port)) transport.connect(username=self.username,password=self.pwd) self.__transport = transport def close(self): self.__transport.close() def upload(self): # 鏈接，上傳 file_name = self.create_file() sftp = paramiko.SFTPClient.from_transport(self.__transport) # 將location.py 上傳至服務器 /tmp/test.py sftp.put(file_name, '/home/wupeiqi/tttttttttttt.py') def rename(self): ssh = paramiko.SSHClient() ssh._transport = self.__transport # 執行命令 stdin, stdout, stderr = ssh.exec_command('mv /home/wupeiqi/tttttttttttt.py /home/wupeiqi/ooooooooo.py') # 獲取命令結果 result = stdout.read() ha = Haproxy() ha.run() Demo
15. Paramiko之肆意妄爲
16. 堡壘機的實現流程介紹
17. 堡壘機的實現
相關文章
- 1. python（十四）網絡編程
- 2. Python網絡編程（四）：UDP編程
- 3. 網絡編程（四）
- 4. python網絡編程——UDP網絡編程
- 5. python-網絡安全編程第四天（數據庫編程&網絡編程）
- 6. Python網絡編程
- 7. python 網絡編程
- 8. python-網絡編程
- 9. python ==》 網絡編程
- 10. Python-網絡編程
- 更多相關文章...
- • netwox網絡工具集入門教程 - TCP/IP教程
- • Rust 併發編程 - RUST 教程
- • RxJava操作符（四）Combining
- • Java 8 Stream 教程
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
相關文章
- 1. python（十四）網絡編程
- 2. Python網絡編程（四）：UDP編程
- 3. 網絡編程（四）
- 4. python網絡編程——UDP網絡編程
- 5. python-網絡安全編程第四天（數據庫編程&網絡編程）
- 6. Python網絡編程
- 7. python 網絡編程
- 8. python-網絡編程
- 9. python ==》 網絡編程
- 10. Python-網絡編程