四. 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-网络编程