【Python第九篇】异步IO数据库队列缓存

本节内容

  1. SelectPollEpoll异步IO与事件驱动
  2. Paramiko SSH
  3. RabbitMQ队列
  4. Redis缓存
  5. pymsql操作
  6. SQLAlchemy

SelectPollEpoll异步IO

参考：http://www.cnblogs.com/alex3714/p/4372426.html　

番外篇 http://www.cnblogs.com/alex3714/articles/5876749.html

select 多并发socket 例子

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#  Author fuyf

import select,socket,sys,queue

server = socket.socket()
server.setblocking(0)

server_addr = ('localhost',9999)

print('starting up on %s port %s' % server_addr)
server.bind(server_addr)

server.listen(5)


inputs = [server, ] #自己也要监测呀,因为server本身也是个fd
outputs = []

message_queues = {}

while True:
    print("waiting for next event...")

    readable, writeable, exeptional = select.select(inputs,outputs,inputs) #如果没有任何fd就绪,那程序就会一直阻塞在这里

    for s in readable: #每个s就是一个socket

        if s is server: #别忘记,上面我们server自己也当做一个fd放在了inputs列表里,传给了select,如果这个s是server,代表server这个fd就绪了,
            #就是有活动了, 什么情况下它才有活动? 当然 是有新连接进来的时候 呀
            #新连接进来了,接受这个连接
            conn, client_addr = s.accept()
            print("new connection from",client_addr)
            conn.setblocking(0)
            inputs.append(conn) #为了不阻塞整个程序,我们不会立刻在这里开始接收客户端发来的数据, 把它放到inputs里, 下一次loop时,这个新连接
            #就会被交给select去监听,如果这个连接的客户端发来了数据 ,那这个连接的fd在server端就会变成就续的,select就会把这个连接返回,返回到
            #readable 列表里,然后你就可以loop readable列表,取出这个连接,开始接收数据了, 下面就是这么干 的

            message_queues[conn] = queue.Queue() #接收到客户端的数据后,不立刻返回 ,暂存在队列里,以后发送

        else: #s不是server的话,那就只能是一个 与客户端建立的连接的fd了
            #客户端的数据过来了,在这接收
            data = s.recv(1024)
            if data:
                print("收到来自[%s]的数据:" % s.getpeername()[0], data)
                message_queues[s].put(data) #收到的数据先放到queue里,一会返回给客户端
                if s not  in outputs:
                    outputs.append(s) #为了不影响处理与其它客户端的连接 , 这里不立刻返回数据给客户端


            else:#如果收不到data代表什么呢? 代表客户端断开了呀
                print("客户端断开了",s)

                if s in outputs:
                    outputs.remove(s) #清理已断开的连接

                inputs.remove(s) #清理已断开的连接

                del message_queues[s] ##清理已断开的连接


    for s in writeable:
        try :
            next_msg = message_queues[s].get_nowait()

        except queue.Empty:
            print("client [%s]" %s.getpeername()[0], "queue is empty..")
            outputs.remove(s)

        else:
            print("sending msg to [%s]"%s.getpeername()[0], next_msg)
            s.send(next_msg.upper())


    for s in exeptional:
        print("handling exception for ",s.getpeername())
        inputs.remove(s)
        if s in outputs:
            outputs.remove(s)
        s.close()

        del message_queues[s]
select socket server
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#  Author fuyf

import socket,sys

messages = [ b'This is the message. ',
             b'It will be sent ',
             b'in parts.',
             ]
server_address = ('localhost', 9999)

# Create a TCP/IP socket
socks = [ socket.socket(),
          socket.socket(),]

# Connect the socket to the port where the server is listening
print('connecting to %s port %s' % server_address)
for s in socks:
    s.connect(server_address)

for message in messages:

    # Send messages on both sockets
    for s in socks:
        print('%s: sending "%s"' % (s.getsockname(), message) )
        s.send(message)

    # Read responses on both sockets
    for s in socks:
        data = s.recv(1024)
        print( '%s: received "%s"' % (s.getsockname(), data) )
        if not data:
            print(sys.stderr, 'closing socket', s.getsockname() )
select socket client.py

selectors模块

import selectors
import socket
 
sel = selectors.DefaultSelector()
 
def accept(sock, mask):
    conn, addr = sock.accept()  # Should be ready
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)
 
def read(conn, mask):
    data = conn.recv(1000)  # Should be ready
    if data:
        print('echoing', repr(data), 'to', conn)
        conn.send(data)  # Hope it won't block
    else:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()
 
sock = socket.socket()
sock.bind(('localhost', 10000))
sock.listen(100)
sock.setblocking(False)
sel.register(sock, selectors.EVENT_READ, accept)
 
while True:
    events = sel.select()
    for key, mask in events:
        callback = key.data
        callback(key.fileobj, mask)
View Code

Paramiko模块

Python的paramiko模块，该模块机遇SSH用于连接远程服务器并执行相关操作

SSHClient

用于连接远程服务器并执行基本命令

基于用户名密码连接：

import paramiko

# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='192.168.69.105', port=22, username='root', password='servyou')

# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')
# 获取命令结果
result = stdout.read().decode()

print(result)
# 关闭连接
ssh.close()
import paramiko

transport = paramiko.Transport(('192.168.69.105', 22))
transport.connect(username='root', password='servyou')

ssh = paramiko.SSHClient()
ssh._transport = transport

stdin, stdout, stderr = ssh.exec_command('df')
print(stdout.read().decode())

transport.close()
SSHClient 封装 Transport

基于公钥密钥连接：

import paramiko
 
private_key = paramiko.RSAKey.from_private_key_file('id_rsa.txt')
 
# 创建SSH对象
ssh = paramiko.SSHClient()
# 允许连接不在know_hosts文件中的主机
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
# 连接服务器
ssh.connect(hostname='192.168.69.105', port=22, username='root', key=private_key)
 
# 执行命令
stdin, stdout, stderr = ssh.exec_command('df')

stdin, stdout, stderr = ssh.exec_command('df;ifconfig') # 多个命令用;隔开
# 获取命令结果
res,err = stdout.read().decode(),stderr.read().decode()
result = res if res else err  # 判断res和err哪个有输出
 
# 关闭连接
ssh.close()

import paramiko

private_key = paramiko.RSAKey.from_private_key_file('id_rsa.txt')

transport = paramiko.Transport(('192.168.69.105', 22))
transport.connect(username='root', pkey=private_key)

ssh = paramiko.SSHClient()
ssh._transport = transport

stdin, stdout, stderr = ssh.exec_command('df')
result = stdout.read().decode()
transport.close()
SSHClient 封装 Transport

SFTPClient

用于连接远程服务器并执行上传下载

基于用户名密码上传下载

import paramiko
 
transport = paramiko.Transport(('192.168.69.105',22))
transport.connect(username='root',password='servyou')

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('id_rsa.txt')
 
transport = paramiko.Transport(('192.168.69.105', 22))
transport.connect(username='root', 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()
View Code

RabbitMQ队列

    RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。
    MQ全称为Message Queue, 消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

一、安装RabbitMQ

windows:

先安装Erlang，在安装RabbitMQ，安装完成后会在生成系统服务，并确保在启动状态

如果启动报错：产生pika.exceptions.ConnectionClosed错误，是因为没有开启rabbitmq服务。

Linux：

# RabbitMQ服务端
rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm  # 安装配置epel源
yum -y install erlang                  　  # 安装erlang
yum -y install rabbitmq-server        # 安装RabbitMQ

# RabbitMQ的API
pip install pika
or
easy_install pika
or
源码
https://pypi.python.org/pypi/pika

二、RabbitMQ应用场景

    RabbitMQ是一个消息代理，从“生产者”接收消息并传递消息至“消费者”，期间可根据规则路由、缓存、持久化消息。“生产者”也即message发送者以下简称P，相对应的“消费者”乃message接收者以下简称C，message通过queue由P到C，queue存在于RabbitMQ，可存储尽可能多的message，多个P可向同一queue发送message，多个C可从同一个queue接收message。

应用场景1-“Hello Word”

一个P向queue发送一个message，一个C从该queue接收message并打印。

producer，连接至RabbitMQ Server，声明队列，发送message，关闭连接，退出。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#  Author fuyf

import pika

# 与RabbitMQ Server建立连接
# 连接到的broker在本机-localhost上
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 声明队列以向其发送消息消息
# 向不存在的位置发送消息时RabbitMQ将消息丢弃
# queue='hello'指定队列名字
channel.queue_declare(queue='hello')

# message不能直接发送给queue，需经exchange到达queue，此处使用以空字符串标识的默认的exchange
# 使用默认exchange时允许通过routing_key明确指定message将被发送给哪个queue
# body参数指定了要发送的message内容
channel.basic_publish(exchange='',routing_key='hello',body='Hello World!')

print(" [x] Sent 'Hello World!'")
# 关闭与RabbitMq Server间的连接
connection.close()
send.py

consumer，连接至RabbitMQ Server，声明队列，接收消息并进行处理这里为打印出消息，退出。

import pika

# 建立到达RabbitMQ Server的connection
# 此处RabbitMQ Server位于本机-localhost
connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# 声明queue，确认要从中接收message的queue
# queue_declare函数是幂等的，可运行多次，但只会创建一次
# 若可以确信queue是已存在的，则此处可省略该声明，如producer已经生成了该queue
# 但在producer和consumer中重复声明queue是一个好的习惯
channel.queue_declare(queue='hello')
# 定义回调函数
# 一旦从queue中接收到一个message回调函数将被调用
# ch：channel
# method：
# properties：
# body：message
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

# 从queue接收message的参数设置
# 包括从哪个queue接收message，用于处理message的callback，是否要确认message
# 默认情况下是要对消息进行确认的，以防止消息丢失。
# 此处将no_ack明确指明为True，不对消息进行确认。
channel.basic_consume(callback,queue='hello',no_ack=True)

print(' [*] Waiting for messages. To exit press CTRL+C')
# 开始循环从queue中接收message并使用callback进行处理
channel.start_consuming()
receive.py

应用场景2-work queues

将耗时的消息处理通过队列分配给多个consumer来处理，我们称此处的consumer为worker，我们将此处的queue称为Task Queue，其目的是为了避免资源密集型的task的同步处理，也即立即处理task并等待完成。相反，调度task使其稍后被处理。也即把task封装进message并发送到task queue，worker进程在后台运行，从task queue取出task并执行job，若运行了多个worker，则task可在多个worker间分配。


建立连接，声明队列，发送可以模拟耗时任务的message，断开连接、退出。

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 仅仅对message进行确认不能保证message不丢失，比如RabbitMQ崩溃了queue就会丢失
# 因此还需使用durable=True声明queue是持久化的，这样即便Rabb崩溃了重启后queue仍然存在
channel.queue_declare(queue='task_queue', durable=True)

# 从命令行构造将要发送的message
message = ' '.join(sys.argv[1:]) or "Hello World!"

# 除了要声明queue是持久化的外，还需声明message是持久化的
# basic_publish的properties参数指定message的属性
# 此处pika.BasicProperties中的delivery_mode=2指明message为持久的
# 这样一来RabbitMQ崩溃重启后queue仍然存在其中的message也仍然存在
channel.basic_publish(exchange='',
                      routing_key='task_queue',
                      body=message,
                      properties=pika.BasicProperties(
                          delivery_mode=2,  # 确保消息持久化
                      ))
print(" [x] Sent %r" % (message,))
connection.close()
new_task.py

建立连接，声明队列，不断的接收message，处理任务，进行确认。

import pika
import time

# 默认情况RabbitMQ将message以round-robin方式发送给下一个consumer
# 每个consumer接收到的平均message量是一样的
# 可以同时运行两个或三个该程序进行测试

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='localhost'))
channel = connection.channel()

# 仅仅对message进行确认不能保证message不丢失，比如RabbitMQ崩溃了
# 还需使用durable=True声明queue是持久化的，这样即便Rabb崩溃了重启后queue仍然存在其中的message不会丢失
# RabbitMQ中不允许使用不同的参数定义同名queue
channel.queue_declare(queue='task_queue', durable=True)

# 回调函数，函数体模拟耗时的任务处理
def callback(ch, method, properties, body):
    print(" [x] Received %r" % (body,))
    time.sleep(20)
    print(" [x] Done")
    # 对message进行确认
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 若存在多个consumer每个consumer的负载可能不同，有些处理的快有些处理的慢
# RabbitMQ并不管这些，只是简单的以round-robin的方式分配message
# 这可能造成某些consumer积压很多任务处理不完而一些consumer长期处于饥饿状态
# 可以使用prefetch_count=1的basic_qos方法可告知RabbitMQ只有在consumer处理并确认了上一个message后才分配新的message给他
# 否则分给另一个空闲的consumer
channel.basic_qos(prefetch_count=1)
# 这里移除了no_ack=True这个参数，也即需要对message进行确认（默认行为）
# 否则consumer在偶然down后其正在处理和分配到该consumer还未处理的message可能发生丢失
# 因为此时RabbitMQ在发送完message后立即从内存删除该message
# 假如没有设置no_ack=True则consumer在偶然down掉后其正在处理和分配至该consumer但还未来得及处理的message会重新分配到其他consumer
# 没有设置no_ack=True则consumer在收到message后会向RabbitMQ反馈已收到并处理了message告诉RabbitMQ可以删除该message
# RabbitMQ中没有超时的概念，只有在consumer down掉后重新分发message
channel.basic_consume(callback,
                      queue='task_queue')

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
worker.py

参数设置解释：

  • durable=True:指定durable参数为真，队列将持久化；

  • properties=pika.BasicProperties(delivery_mode = 2,): #启用消息持久化，可以防止RabbitMQ Server 重启或者crash引起的数据丢失。

  • ch.basic_ack(delivery_tag = method.delivery_tag):告诉rabbitmq消息已经正确处理。如果没有这条代码，Consumer退出时，Message会重新分发。然后RabbitMQ会占用越来越多的内存，由于RabbitMQ会长时间运行，因此这个“内存泄漏”是致命的。

  • no_ack=True:默认为假。设置no-ack参数为真，每次Consumer接到数据后，而不管是否处理完成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。

  • channel.basic_qos(prefetch_count=1)公平分发。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。

应用场景3-Publish/Subscribe（消息发布/订阅）

在应用场景2中一个message(task)仅被传递给了一个comsumer(worker)。现在我们设法将一个message传递给多个consumer。这种模式被称为publish/subscribe。此处以一个简单的日志系统为例进行说明。该系统包含一个log发送程序和一个log接收并打印的程序。由log发送者发送到queue的消息可以被所有运行的log接收者接收。因此，我们可以运行一个log接收者直接在屏幕上显示log，同时运行另一个log接收者将log写入磁盘文件。类似广播的效果。

fanout: 所有bind到此exchange的queue都可以接收消息。

日志消息发送者：建立连接，声明fanout类型的exchange，通过exchage向queue发送日志消息，消息被广播给所有接收者，关闭连接，退出。

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()
# producer只能通过exchange将message发给queue
# exchange的类型决定将message路由至哪些queue
# 可用的exchange类型：direct	opicheadersfanout
# 此处定义一个名称为'logs'的'fanout'类型的exchange，'fanout'类型的exchange简单的将message广播到它所知道的所有queue
channel.exchange_declare(exchange='logs',type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"
# 将message publish到名为log的exchange中
# 因为是fanout类型的exchange，这里无需指定routing_key
channel.basic_publish(exchange='logs',
                      routing_key='',
                      body=message)

print(" [x] Sent %r" % (message,))

connection.close()
生产者

日志消息接收者：建立连接，声明exchange，将exchange与queue进行绑定，开始不停的接收log并打印。

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
channel = connection.channel()

# 作为好的习惯，在producer和consumer中分别声明一次以保证所要使用的exchange存在
channel.exchange_declare(exchange='logs',type='fanout')

# 在不同的producer和consumer间共享queue时指明queue的name是重要的
# 但某些时候，比如日志系统，需要接收所有的log message而非一个子集
# 而且仅对当前的message 流感兴趣，对于过时的message不感兴趣，那么
# 可以申请一个临时队列这样，每次连接到RabbitMQ时会以一个随机的名字生成
# 一个新的空的queue，将exclusive置为True，这样在consumer从RabbitMQ断开后会删除该queue
result = channel.queue_declare(exclusive=True)
# 用于获取临时queue的name
queue_name = result.method.queue
# exchange与queue之间的关系成为binding
# binding告诉exchange将message发送该哪些queue
channel.queue_bind(exchange='logs',queue=queue_name)

def callback(ch, method, properties, body):
    print(" [x] %r" % (body,))

# 从指定地queue中consume message且不确认
channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

print(' [*] Waiting for logs. To exit press CTRL+C')
channel.start_consuming()
消费者

应用场景4-有选择的接收消息(exchange type=direct)　

direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息

RabbitMQ还支持根据关键字发送，即：队列绑定关键字，发送者将数据根据关键字发送到消息exchange，exchange根据 关键字判定应该将数据发送至指定队列。

应用场景3中构建了简单的log系统，可以将log message广播至多个receiver。现在我们将考虑只把指定的message类型发送给其subscriber，比如只把error message写到log file而将所有log message显示在控制台。

log message发送者：建立连接，声明direct类型的exchange，生成并发送log message到exchange，关闭连接，退出。

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='localhost'))
channel = connection.channel()

# 声明一个名为direct_logs的direct类型的exchange
# direct类型的exchange
channel.exchange_declare(exchange='direct_logs',
                         type='direct')

# 从命令行获取basic_publish的配置参数
severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
message = ' '.join(sys.argv[2:]) or 'Hello World!'

# 向名为direct_logs的exchage按照设置的routing_key发送message
channel.basic_publish(exchange='direct_logs',
                      routing_key=severity,
                      body=message)

print(" [x] Sent %r:%r" % (severity, message))
connection.close()
生产者

log message接收者：建立连接，声明direct类型的exchange，声明queue，使用提供的参数作为routing_key将queue绑定到exchange，开始循环接收log message并打印。

import pika
import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='localhost'))
channel = connection.channel()

# 声明一个名为direct_logs类型为direct的exchange
# 同时在producer和consumer中声明exchage或queue是个好习惯，以保证其存在
channel.exchange_declare(exchange='direct_logs',
                         type='direct')

result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue

# 从命令行获取参数:routing_key
severities = sys.argv[1:]
if not severities:
    sys.stderr.write("Usage: %s [info] [warning] [error]
" % sys.argv[0])
    sys.exit(1)

for severity in severities:
    # exchange和queue之间的binding可接受routing_key参数
    # 该参数的意义依赖于exchange的类型
    # fanout类型的exchange直接忽略该参数
    # direct类型的exchange精确匹配该关键字进行message路由
    # 对多个queue使用相同的binding_key是合法的
    channel.queue_bind(exchange='direct_logs',
                       queue=queue_name,
                       routing_key=severity)

def callback(ch, method, properties, body):
    print(" [x] %r:%r" % (method.routing_key, body,))

channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)

print(' [*] Waiting for logs. To exit press CTRL+C')
channel.start_consuming()
消费者

应用场景5-topic

topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

应用场景4中改进的log系统中用direct类型的exchange替换应用场景3中的fanout类型exchange实现将不同的log message发送给不同的subscriber（也即分别通过不同的routing_key将queue绑定到exchange，这样exchange便可将不同的message根据message内容路由至不同的queue）。但仍然存在限制，不能根据多个规则路由消息，比如接收者要么只能收error类型的log message要么只能收info类型的message。如果我们不仅想根据log的重要级别如info、warning、error等来进行log message路由还想同时根据log message的来源如auth、cron、kern来进行路由。为了达到此目的，需要topic类型的exchange。topic类型的exchange中routing_key中可以包含两个特殊字符：“*”用于替代正好一个词，“#”用于0个或多个词。

log message接收者：建立连接，声明topic类型的exchange,声明queue，根据程序参数构造routing_key，根据routing_key将queue绑定到exchange，循环接收并处理message。

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]...
" % 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()
消费者

log message发送者：建立连接、声明topic类型的exchange、根据程序参数构建routing_key和要发送的message，以构建的routing_key将message发送给topic类型的exchange，关闭连接，退出。

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

应用场景6-RPC

在应用场景2中描述了如何使用work queue将耗时的task分配到不同的worker中。但是，如果我们task是想在远程的计算机上运行一个函数并等待返回结果呢。这根场景2中的描述是一个完全不同的故事。这一模式被称为远程过程调用。现在，我们将构建一个RPC系统，包含一个client和可扩展的RPC server，通过返回斐波那契数来模拟RPC service。

RPC server：建立连接，声明queue，定义了一个返回指定数字的斐波那契数的函数，定义了一个回调函数在接收到包含参数的调用请求后调用自己的返回斐波那契数的函数并将结果发送到与接收到message的queue相关联的queue，并进行确认。开始接收调用请求并用回调函数进行请求处理。

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters(
    host='localhost'))

channel = connection.channel()

channel.queue_declare(queue='rpc_queue')


def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)


def on_request(ch, method, props, body):
    n = int(body)

    print(" [.] fib(%s)" % n)
    response = fib(n)

    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id= 
                                                         props.correlation_id),
                     body=str(response))
    ch.basic_ack(delivery_tag=method.delivery_tag)


channel.basic_qos(prefetch_count=1)
channel.basic_consume(on_request, queue='rpc_queue')

print(" [x] Awaiting RPC requests")
channel.start_consuming()
rpc_server.py

RPC client：远程过程调用发起者：定义了一个类，类中初始化到RabbitMQ Server的连接、声明回调queue、开始在回调queue上等待接收响应、定义了在回调queue上接收到响应后的处理函数on_response根据响应关联的correlation_id属性作出响应、定义了调用函数并在其中向调用queue发送包含correlation_id等属性的调用请求、初始化一个client实例，以30为参数发起远程过程调用。

import pika
import uuid

class FibonacciRpcClient(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))

        self.channel = self.connection.channel()

        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue

        self.channel.basic_consume(self.on_response, no_ack=True,
                                   queue=self.callback_queue)

    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',
                                   properties=pika.BasicProperties(
                                       reply_to=self.callback_queue,
                                       correlation_id=self.corr_id,
                                   ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()
        return int(self.response)


fibonacci_rpc = FibonacciRpcClient()

print(" [x] Requesting fib(7)")
response = fibonacci_rpc.call(7)
print(" [.] Got %r" % response)
rpc_client.py

RabbitMQ web管理工具

默认安装的Rabbit MQ 监听端口是5672。

使用Rabbit MQ 管理插件，可以更好的可视化方式查看Rabbit MQ 服务器实例的状态：先定位到rabbitmq安装目录， 然后输入命令rabbitmq-plugins enable rabbitmq_management。启动后直接在浏览器地址输入：http://localhost:15672/ 账号密码都是：guest 即可。

Redis

redis是一个key-value存储系统。和Memcached类似，它支持存储的value类型相对更多，包括string(字符串)、list(链表)、set(集合)、zset(sorted set --有序集合)和hash（哈希类型）。这些数据类型都支持push/pop、add/remove及取交集并集和差集及更丰富的操作，而且这些操作都是原子性的。在此基础上，redis支持各种不同方式的排序。与memcached一样，为了保证效率，数据都是缓存在内存中。区别的是redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件，并且在此基础上实现了master-slave(主从)同步。

一、Redis安装和基本使用

1
2
3
4
wget http://download.redis.io/releases/redis-3.0.6.tar.gz
tar xzf redis-3.0.6.tar.gz
cd redis-3.0.6
make

启动服务端

1
src/redis-server

启动客户端

1
2
3
4
5
src/redis-cli
redis> set foo bar
OK
redis> get foo
"bar"

二、Python操作Redis

1
2
3
4
5
6
7
sudo pip install redis
or
sudo easy_install redis
or
源码安装
详见：https://github.com/WoLpH/redis-py

API使用

redis-py 的API的使用可以分类为：

  • 连接方式
  • 连接池
  • 操作
    • String 操作
    • Hash 操作
    • List 操作
    • Set 操作
    • Sort Set 操作
  • 管道
  • 发布订阅

1、连接方式

redis-py提供两个类Redis和StrictRedis用于实现Redis的命令，StrictRedis用于实现大部分官方的命令，并使用官方的语法和命令，Redis是StrictRedis的子类，用于向后兼容旧版本的redis-py。

import redis

r = redis.Redis(host='192.168.69.102', port=6379)
r.set('foo', 'Bar')
print(r.get('foo').decode())

2、连接池

redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池。 

import redis
pool = redis.ConnectionPool(host='192.168.69.102', port=6379)
r = redis.Redis(connection_pool=pool)
r.set('name','chris')
print(r.get('name').decode())

3、操作

  • String操作，redis中的String在在内存中按照一个name对应一个value来存储。

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

1
2
3
4
5
6
在Redis中设置值，默认，不存在则创建，存在则修改
参数：
     ex，过期时间（秒）
     px，过期时间（毫秒）
     nx，如果设置为True，则只有name不存在时，当前set操作才执行
     xx，如果设置为True，则只有name存在时，岗前set操作才执行

setnx(name, value)

1
设置值，只有name不存在时，执行设置操作（添加）

setex(name, value, time)

1
2
3
# 设置值
# 参数：
    # time，过期时间（数字秒 或 timedelta对象）

psetex(name, time_ms, value)

1
2
3
# 设置值
# 参数：
    # time_ms，过期时间（数字毫秒 或 timedelta对象）

mset(*args, **kwargs)

1
2
3
4
5
批量设置值
如：
    mset(k1='v1', k2='v2')
    或
    mget({'k1': 'v1', 'k2': 'v2'})

get(name)

1
获取值

mget(keys, *args)

1
2
3
4
5
批量获取
如：
    mget('ylr', 'wupeiqi')
    或
    r.mget(['ylr', 'wupeiqi'])

getset(name, value)

1
设置新值并获取原来的值

getrange(key, start, end)

1
2
3
4
5
6
# 获取子序列（根据字节获取，非字符）
# 参数：
    # name，Redis 的 name
    # start，起始位置（字节）
    # end，结束位置（字节）
# 如： "武沛齐" ，0-3表示 "武"

setrange(name, offset, value)

1
2
3
4
# 修改字符串内容，从指定字符串索引开始向后替换（新值太长时，则向后添加）
# 参数：
    # offset，字符串的索引，字节（一个汉字三个字节）
    # value，要设置的值

setbit(name, offset, value)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
# 对name对应值的二进制表示的位进行操作
 
# 参数：
    # name，redis的name
    # offset，位的索引（将值变换成二进制后再进行索引）
    # value，值只能是 1 或 0
 
# 注：如果在Redis中有一个对应： n1 = "foo"，
        那么字符串foo的二进制表示为：01100110 01101111 01101111
    所以，如果执行 setbit('n1', 7, 1)，则就会将第7位设置为1，
        那么最终二进制则变成 01100111 01101111 01101111，即："goo"
 
# 扩展，转换二进制表示：
 
    # source = "武沛齐"
    source = "foo"
 
    for i in source:
        num = ord(i)
        print bin(num).replace('b','')
 
    特别的，如果source是汉字 "武沛齐"怎么办？
    答：对于utf-8，每一个汉字占 3 个字节，那么 "武沛齐" 则有 9个字节
       对于汉字，for循环时候会按照 字节 迭代，那么在迭代时，将每一个字节转换 十进制数，然后再将十进制数转换成二进制
        11100110 10101101 10100110 11100110 10110010 10011011 11101001 10111101 10010000
        -------------------------- ----------------------------- -----------------------------
                    武                         沛                           齐

getbit(name, offset)

1
# 获取name对应的值的二进制表示中的某位的值 （0或1）

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

1
2
3
4
5
# 获取name对应的值的二进制表示中 1 的个数
# 参数：
    # key，Redis的name
    # start，位起始位置
    # end，位结束位置

bitop(operation, dest, *keys)

1
2
3
4
5
6
7
8
9
10
# 获取多个值，并将值做位运算，将最后的结果保存至新的name对应的值
 
# 参数：
    # operation,AND（并） 、 OR（或） 、 NOT（非） 、 XOR（异或）
    # dest, 新的Redis的name
    # *keys,要查找的Redis的name
 
# 如：
    bitop("AND", 'new_name', 'n1', 'n2', 'n3')
    # 获取Redis中n1,n2,n3对应的值，然后讲所有的值做位运算（求并集），然后将结果保存 new_name 对应的值中

strlen(name)

1
# 返回name对应值的字节长度（一个汉字3个字节）

incr(self, name, amount=1)

1
2
3
4
5
6
7
# 自增 name对应的值，当name不存在时，则创建name＝amount，否则，则自增。
 
# 参数：
    # name,Redis的name
    # amount,自增数（必须是整数）
 
# 注：同incrby

incrbyfloat(self, name, amount=1.0)

1
2
3
4
5
# 自增 name对应的值，当name不存在时，则创建name＝amount，否则，则自增。
 
# 参数：
    # name,Redis的name
    # amount,自增数（浮点型）

decr(self, name, amount=1)

1
2
3
4
5
# 自减 name对应的值，当name不存在时，则创建name＝amount，否则，则自减。
 
# 参数：
    # name,Redis的name
    # amount,自减数（整数）

append(key, value)

1
2
3
4
5
# 在redis name对应的值后面追加内容
 
# 参数：
    key, redis的name
    value, 要追加的字符串
  • Hash操作，redis中Hash在内存中的存储格式如下图：

hset(name, key, value)

1
2
3
4
5
6
7
8
9
# name对应的hash中设置一个键值对（不存在，则创建；否则，修改）
 
# 参数：
    # name，redis的name
    # key，name对应的hash中的key
    # value，name对应的hash中的value
 
# 注：
    # hsetnx(name, key, value),当name对应的hash中不存在当前key时则创建（相当于添加）

hmset(name, mapping)

1
2
3
4
5
6
7
8
# 在name对应的hash中批量设置键值对
 
# 参数：
    # name，redis的name
    # mapping，字典，如：{'k1':'v1', 'k2': 'v2'}
 
# 如：
    # r.hmset('xx', {'k1':'v1', 'k2': 'v2'})

hget(name,key)

1
# 在name对应的hash中获取根据key获取value

hmget(name, keys, *args)

1
2
3
4
5
6
7
8
9
10
11
# 在name对应的hash中获取多个key的值
 
# 参数：
    # name，reids对应的name
    # keys，要获取key集合，如：['k1', 'k2', 'k3']
    # *args，要获取的key，如：k1,k2,k3
 
# 如：
    # r.mget('xx', ['k1', 'k2'])
    # 或
    # print r.hmget('xx', 'k1', 'k2')

hgetall(name)

1
获取name对应hash的所有键值

hlen(name)

1
# 获取name对应的hash中键值对的个数

hkeys(name)

1
# 获取name对应的hash中所有的key的值

hvals(name)

1
# 获取name对应的hash中所有的value的值

hexists(name, key)

1
# 检查name对应的hash是否存在当前传入的key

hdel(name,*keys)

1
# 将name对应的hash中指定key的键值对删除

hincrby(name, key, amount=1)

1
2
3
4
5
# 自增name对应的hash中的指定key的值，不存在则创建key=amount
# 参数：
    # name，redis中的name
    # key， hash对应的key
    # amount，自增数（整数）

hincrbyfloat(name, key, amount=1.0)

1
2
3
4
5
6
7
8
# 自增name对应的hash中的指定key的值，不存在则创建key=amount
 
# 参数：
    # name，redis中的name
    # key， hash对应的key
    # amount，自增数（浮点数）
 
# 自增name对应的hash中的指定key的值，不存在则创建key=amount

hscan(name, cursor=0, match=None, count=None)

1
2
3
4
5
6
7
8
9
10
11
12
13
# 增量式迭代获取，对于数据大的数据非常有用，hscan可以实现分片的获取数据，并非一次性将数据全部获取完，从而放置内存被撑爆
 
# 参数：
    # name，redis的name
    # cursor，游标（基于游标分批取获取数据）
    # match，匹配指定key，默认None 表示所有的key
    # count，每次分片最少获取个数，默认None表示采用Redis的默认分片个数
 
# 如：
    # 第一次：cursor1, data1 = r.hscan('xx', cursor=0, match=None, count=None)
    # 第二次：cursor2, data1 = r.hscan('xx', cursor=cursor1, match=None, count=None)
    # ...
    # 直到返回值cursor的值为0时，表示数据已经通过分片获取完毕

hscan_iter(name, match=None, count=None)

1
2
3
4
5
6
7
8
9
# 利用yield封装hscan创建生成器，实现分批去redis中获取数据
 
# 参数：
    # match，匹配指定key，默认None 表示所有的key
    # count，每次分片最少获取个数，默认None表示采用Redis的默认分片个数
 
# 如：
    # for item in r.hscan_iter('xx'):
    #     print item
  • List操作，redis中的List在在内存中按照一个name对应一个List来存储。如图：

lpush(name,values)

1
2
3
4
5
6
7
8
# 在name对应的list中添加元素，每个新的元素都添加到列表的最左边
 
# 如：
    # r.lpush('oo', 11,22,33)
    # 保存顺序为: 33,22,11
 
# 扩展：
    # rpush(name, values) 表示从右向左操作

lpushx(name,value)

1
2
3
4
# 在name对应的list中添加元素，只有name已经存在时，值添加到列表的最左边
 
# 更多：
    # rpushx(name, value) 表示从右向左操作

llen(name)

1
# name对应的list元素的个数

linsert(name, where, refvalue, value))

1
2
3
4
5
6
7
# 在name对应的列表的某一个值前或后插入一个新值
 
# 参数：
    # name，redis的name
    # where，BEFORE或AFTER
    # refvalue，标杆值，即：在它前后插入数据
    # value，要插入的数据

r.lset(name, index, value)

1
2
3
4
5
6
# 对name对应的list中的某一个索引位置重新赋值
 
# 参数：
    # name，redis的name
    # index，list的索引位置
    # value，要设置的值

r.lrem(name, value, num)

1
2
3
4
5
6
7
8
# 在name对应的list中删除指定的值
 
# 参数：
    # name，redis的name
    # value，要删除的值
    # num，  num=0，删除列表中所有的指定值；
           # num=2,从前到后，删除2个；
           # num=-2,从后向前，删除2个

lpop(name)

1
2
3
4
# 在name对应的列表的左侧获取第一个元素并在列表中移除，返回值则是第一个元素
 
# 更多：
    # rpop(name) 表示从右向左操作

lindex(name, index)

1
在name对应的列表中根据索引获取列表元素

lrange(name, start, end)

1
2
3
4
5
# 在name对应的列表分片获取数据
# 参数：
    # name，redis的name
    # start，索引的起始位置
    # end，索引结束位置

ltrim(name, start, end)

1
2
3
4
5
# 在name对应的列表中移除没有在start-end索引之间的值
# 参数：
    # name，redis的name
    # start，索引的起始位置
    # end，索引结束位置

rpoplpush(src, dst)

1
2
3
4
# 从一个列表取出最右边的元素，同时将其添加至另一个列表的最左边
# 参数：
    # src，要取数据的列表的name
    # dst，要添加数据的列表的name

blpop(keys, timeout)

1
2
3
4
5
6
7
8
# 将多个列表排列，按照从左到右去pop对应列表的元素
 
# 参数：
    # keys，redis的name的集合
    # timeout，超时时间，当元素所有列表的元素获取完之后，阻塞等待列表内有数据的时间（秒）, 0 表示永远阻塞
 
# 更多：
    # r.brpop(keys, timeout)，从右向左获取数据

brpoplpush(src, dst, timeout=0)

1
2
3
4
5
6
# 从一个列表的右侧移除一个元素并将其添加到另一个列表的左侧
 
# 参数：
    # src，取出并要移除元素的列表对应的name
    # dst，要插入元素的列表对应的name
    # timeout，当src对应的列表中没有数据时，阻塞等待其有数据的超时时间（秒），0 表示永远阻塞

自定义增量迭代

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 由于redis类库中没有提供对列表元素的增量迭代，如果想要循环name对应的列表的所有元素，那么就需要：
    # 1、获取name对应的所有列表
    # 2、循环列表
# 但是，如果列表非常大，那么就有可能在第一步时就将程序的内容撑爆，所有有必要自定义一个增量迭代的功能：
 
def list_iter(name):
    """
    自定义redis列表增量迭代
    :param name: redis中的name，即：迭代name对应的列表
    :return: yield 返回 列表元素
    """
    list_count = r.llen(name)
    for index in xrange(list_count):
        yield r.lindex(name, index)
 
# 使用
for item in list_iter('pp'):
    print item
  • Set操作，Set集合就是不允许重复的列表

sadd(name,values)

1
# name对应的集合中添加元素

scard(name)

1
获取name对应的集合中元素个数

sdiff(keys, *args)

1
在第一个name对应的集合中且不在其他name对应的集合的元素集合

sdiffstore(dest, keys, *args)

1
# 获取第一个name对应的集合中且不在其他name对应的集合，再将其新加入到dest对应的集合中

sinter(keys, *args)

1
# 获取多一个name对应集合的并集

sinterstore(dest, keys, *args)

1
# 获取多一个name对应集合的并集，再讲其加入到dest对应的集合中

sismember(name, value)

1
# 检查value是否是name对应的集合的成员

smembers(name)

1
# 获取name对应的集合的所有成员

smove(src, dst, value)

1
# 将某个成员从一个集合中移动到另外一个集合

spop(name)

1
# 从集合的右侧（尾部）移除一个成员，并将其返回

srandmember(name, numbers)

1
# 从name对应的集合中随机获取 numbers 个元素

srem(name, values)

1
# 在name对应的集合中删除某些值

sunion(keys, *args)

1
# 获取多一个name对应的集合的并集

sunionstore(dest,keys, *args)

1
# 获取多一个name对应的集合的并集，并将结果保存到dest对应的集合中

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

1
# 同字符串的操作，用于增量迭代分批获取元素，避免内存消耗太大
  •  有序集合，在集合的基础上，为每元素排序；元素的排序需要根据另外一个值来进行比较，所以，对于有序集合，每一个元素有两个值，即：值和分数，分数专门用来做排序。

 

zadd(name, *args, **kwargs)

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

zcard(name)

1
# 获取name对应的有序集合元素的数量

zcount(name, min, max)

1
# 获取name对应的有序集合中分数 在 [min,max] 之间的个数

zincrby(name, value, amount)

1
# 自增name对应的有序集合的 name 对应的分数

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 按照索引范围获取name对应的有序集合的元素
 
# 参数：
    # name，redis的name
    # start，有序集合索引起始位置（非分数）
    # end，有序集合索引结束位置（非分数）
    # desc，排序规则，默认按照分数从小到大排序
    # withscores，是否获取元素的分数，默认只获取元素的值
    # score_cast_func，对分数进行数据转换的函数
 
# 更多：
    # 从大到小排序
    # zrevrange(name, start, end, withscores=False, score_cast_func=float)
 
    # 按照分数范围获取name对应的有序集合的元素
    # zrangebyscore(name, min, max, start=None, num=None, withscores=False, score_cast_func=float)
    # 从大到小排序
    # zrevrangebyscore(name, max, min, start=None, num=None, withscores=False, score_cast_func=float)

zrank(name, value)

1
2
3
4
# 获取某个值在 name对应的有序集合中的排行（从 0 开始）
 
# 更多：
    # zrevrank(name, value)，从大到小排序

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 当有序集合的所有成员都具有相同的分值时，有序集合的元素会根据成员的 值 （lexicographical ordering）来进行排序，而这个命令则可以返回给定的有序集合键 key 中， 元素的值介于 min 和 max 之间的成员
# 对集合中的每个成员进行逐个字节的对比（byte-by-byte compare）， 并按照从低到高的顺序， 返回排序后的集合成员。 如果两个字符串有一部分内容是相同的话， 那么命令会认为较长的字符串比较短的字符串要大
 
# 参数：
    # name，redis的name
    # min，左区间（值）。 + 表示正无限； - 表示负无限； ( 表示开区间； [ 则表示闭区间
    # min，右区间（值）
    # start，对结果进行分片处理，索引位置
    # num，对结果进行分片处理，索引后面的num个元素
 
# 如：
    # ZADD myzset 0 aa 0 ba 0 ca 0 da 0 ea 0 fa 0 ga
    # r.zrangebylex('myzset', "-", "[ca") 结果为：['aa', 'ba', 'ca']
 
# 更多：
    # 从大到小排序
    # zrevrangebylex(name, max, min, start=None, num=None)

zrem(name, values)

1
2
3
# 删除name对应的有序集合中值是values的成员
 
# 如：zrem('zz', ['s1', 's2'])

zremrangebyrank(name, min, max)

1
# 根据排行范围删除

zremrangebyscore(name, min, max)

1
# 根据分数范围删除

zremrangebylex(name, min, max)

1
# 根据值返回删除

zscore(name, value)

1
# 获取name对应有序集合中 value 对应的分数

zinterstore(dest, keys, aggregate=None)

1
2
# 获取两个有序集合的交集，如果遇到相同值不同分数，则按照aggregate进行操作
# aggregate的值为:  SUM  MIN  MAX

zunionstore(dest, keys, aggregate=None)

1
2
# 获取两个有序集合的并集，如果遇到相同值不同分数，则按照aggregate进行操作
# aggregate的值为:  SUM  MIN  MAX

zscan(name, cursor=0, match=None, count=None, score_cast_func=float)
zscan_iter(name, match=None, count=None,score_cast_func=float)

1
# 同字符串相似，相较于字符串新增score_cast_func，用来对分数进行操作
  • 其他常用操作

delete(*names)

1
# 根据删除redis中的任意数据类型

exists(name)

1
# 检测redis的name是否存在

keys(pattern='*')

1
2
3
4
5
6
7
# 根据模型获取redis的name
 
# 更多：
    # KEYS * 匹配数据库中所有 key 。
    # KEYS h?llo 匹配 hello ， hallo 和 hxllo 等。
    # KEYS h*llo 匹配 hllo 和 heeeeello 等。
    # KEYS h[ae]llo 匹配 hello 和 hallo ，但不匹配 hillo

expire(name ,time)

1
# 为某个redis的某个name设置超时时间

rename(src, dst)

1
# 对redis的name重命名为

move(name, db))

1
# 将redis的某个值移动到指定的db下

randomkey()

1
# 随机获取一个redis的name（不删除）

type(name)

1
# 获取name对应值的类型

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

1
# 同字符串操作，用于增量迭代获取key

4、管道

redis-py默认在执行每次请求都会创建（连接池申请连接）和断开（归还连接池）一次连接操作，如果想要在一次请求中指定多个命令，则可以使用pipline实现一次请求指定多个命令，并且默认情况下一次pipline 是原子性操作。

import redis,time

pool = redis.ConnectionPool(host='192.168.69.102', port=6379)
r = redis.Redis(connection_pool=pool)
pipe = r.pipeline(transaction=True)

pipe.set('name1', 'chris1')
time.sleep(20)
pipe.set('role1', 'student1')

pipe.execute()

5、发布订阅

略

pymsql操作

Python操作MySQL主要使用两种方式：

  • 原生模块 pymsql
  • ORM框架 SQLAchemy

下载安装

1
pip3 install pymysql

使用操作

1、执行SQL

import pymysql

# 创建连接
conn = pymysql.connect(host='127.0.0.1', port=3306, user='root', passwd='servyou', db='servyoudb')
# 创建游标
cursor = conn.cursor()
# 执行SQL，并返回收影响行数,executemany一次插入多条数据，数据用[]括起来
# effect_row = cursor.executemany("insert into students(name,sex,age)values(%s,%s,%s)",[('chris1','M',22)])

# 执行SQL，并返回收影响行数,插入单条数据
# effect_row1 = cursor.execute("insert into students(name,sex,age)values(%s,%s,%s)",('chris2','M',22))

cursor.execute("select * from students")

# 获取第一行数据
# result  = cursor.fetchone()
# 获取前n行数据
# result  = cursor.fetchmany(3)
# 获取所有数据
result  = cursor.fetchall()
print(result)
#  提交，不然无法保存新建或者修改的数据
conn.commit()
# 关闭游标
cursor.close()
# 关闭连接
conn.close()

SQLAlchemy ORM

ORM英文全称object relational mapping,就是对象映射关系程序，简单来说我们类似python这种面向对象的程序来说一切皆对象，但是我们使用的数据库却都是关系型的，为了保证一致的使用习惯，通过orm将编程语言的对象模型和数据库的关系模型建立映射关系，这样我们在使用编程语言对数据库进行操作的时候可以直接使用编程语言的对象模型进行操作就可以了，而不用直接使用sql语言。

在Python中，最有名的ORM框架是SQLAlchemy。

一、安装sqlalchemy

pip install SQLAlchemy
pip install pymysql    #用pymysql与sqlalchemy交互

二、sqlalchemy使用

filter和filter_by使用总结：
filter 可以像写 sql 的 where 条件那样写 > < 等条件
filter用类名.属性名，比较用==，filter_by直接用属性名，比较用=

filter_by的参数是**kwargs，直接支持组合查询。
filter不支持组合查询，只能连续调用filter来变相实现。
  • 创建表结构
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,INTEGER,String

engine = create_engine("mysql+pymysql://root:servyou@localhost/servyoudb",
                       encoding = 'utf-8',echo = True)
Base = declarative_base()#生成orm基类

class User(Base):
    __tablename__ = 'user' # 表名
    id = Column(INTEGER,primary_key=True)
    name = Column(String(32))
    password = Column(String(64))

Base.metadata.create_all(engine)   #创建表结构
  •  插入数据
# 创建数据
session_class = sessionmaker(bind=engine) #创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
session = session_class()

data1 = User(name='chris1',password='123') #生成要创建的数据对象
data2 = User(name='jack1',password='servyou')

session.add(data1)  # 把要创建的数据对象添加到这个session里，一会统一创建
session.add(data2)

session.commit()  # 提交，创建数据
  •  查询
# 获取第一条数据
result = session.query(User).filter_by(name='chris').first()
print(result.id,result.name,result.password)

# 获取所有数据
result2 = session.query(User.id,User.name).all()
print(result2)

#多条件查询
result3 = session.query(User.id,User.name).filter(User.id>0).filter(User.id<7).all()
print(result3)

print(session.query(User).filter_by(id=1, name='chris_new').all())
  • 其他
# 排序
ret = session.query(User).order_by(User.age.desc()).all()
print(ret)
ret2 = session.query(User).order_by(User.age.desc(), User.id.asc()).all()

# 通配符
# ret = session.query(User).filter(User.name.like('e%')).all()  
ret = session.query(User).filter(~User.name.like('e%')).all()  # 非e%的
print(ret)

#条件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
session.query(User).filter(User.name != 'chris')
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
    or_(
        Users.id < 2,
        and_(Users.name == 'eric', Users.id > 3),
        Users.extra != ""
    )).all()

# 分组
from sqlalchemy.sql import func

ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).all()

ret = session.query(
    func.max(Users.id),
    func.sum(Users.id),
    func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()

# 连表
ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
ret = session.query(Person).join(Favor).all()
ret = session.query(Person).join(Favor, isouter=True).all()

# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()

q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()
View Code
  • 修改
# 先查询出来一条数据，在赋值即可修改
data = session.query(User).filter_by(name='chris').first()
print(data)
data.name = 'chris_new'
session.commit()

session.query(User).filter(User.age > 50).update({"name" : "099"})
session.query(User).filter(User.age > 50).update({User.name: User.name + "099"}, synchronize_session=False)
session.query(User).filter(User.age > 50).update({"age": User.age + 1}, synchronize_session="evaluate")
session.commit()
  • 删除
session.query(Users).filter(Users.id > 2).delete()
session.commit()
  • 回滚
fake_user = User(name='Rain', password='12345')
session.add(fake_user)

session.query(User).filter(User.name.in_(['Jack', 'rain'])).all()  # 这时看session里有你刚添加和修改的数据
session.rollback()  # 此时rollback一下
  • 统计和分组
#统计
print(session.query(User).filter(User.name.like("Ch%")).count())
#分组
from sqlalchemy import func
print(session.query(func.count(User.name),User.name).group_by(User.name).all())
打印原生sql：print(session.query(func.count(User.name),User.name).group_by(User.name))
相当于原生sql为
SELECT count(user.name) AS count_1, user.name AS user_name
FROM user GROUP BY user.name
  • 外键关联
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,INTEGER,String,DATE,ForeignKey
from sqlalchemy.orm import relationship

engine = create_engine("mysql+pymysql://root:servyou@localhost/servyoudb?charset=utf8")
Base = declarative_base()  # 生成orm基类

class Student(Base):
    __tablename__ = 'student' # 表名
    id = Column(INTEGER,primary_key=True)
    name = Column(String(32),nullable=False)
    date = Column(DATE,nullable=False)

    def __repr__(self):
        return '<id:%s name:%s>' %(self.id,self.name)

class  StudyRecord(Base):
    __tablename__ = 'study_record'
    id = Column(INTEGER, primary_key=True)
    day = Column(INTEGER, nullable=False)
    status = Column(String(32), nullable=False)
    stu_id = Column(INTEGER, ForeignKey('student.id'))

    student = relationship("Student", backref="my_record")  # 允许你在user表里通过backref字段反向查出所有它在addresses表里的关联项

    def __repr__(self):
        return "<(%s day:%s status:%s)>" % (self.student.name,self.day,self.status)

Base.metadata.create_all(engine)   #创建表结构
orm_m2.py
from sqlalchemy.orm import sessionmaker
from day8 import orm_m2

session_class = sessionmaker(bind=orm_m2.engine)
session = session_class()

#通过student对象反查关联study_record的记录
ret = session.query(orm_m2.Student).filter(orm_m2.Student.name=='chris').first()
print(ret.my_record)

# s1 = orm_m2.Student(name='chris',date='2017-07-01')
# s2 = orm_m2.Student(name='eric',date='2016-06-01')
# s3 = orm_m2.Student(name='jack',date='2017-07-11')
#
# r1 = orm_m2.StudyRecord(day=1,status='yes',stu_id=1)
# r2 = orm_m2.StudyRecord(day=2,status='no',stu_id=1)
# r3 = orm_m2.StudyRecord(day=3,status='yes',stu_id=1)
# r4 = orm_m2.StudyRecord(day=1,status='yes',stu_id=2)
#
# session.add_all([s1,s2,s3,r1,r2,r3,r4])
session.commit()
orm_api.py
  • 多外键关联
from sqlalchemy import Integer, ForeignKey, String, Column,create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship,sessionmaker

engine = create_engine("mysql+pymysql://root:servyou@localhost/servyoudb?charset=utf8")
Base = declarative_base()  # 生成orm基类

class Customer(Base):
    __tablename__ = 'customer'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))

    billing_address_id = Column(Integer, ForeignKey("address.id"))
    shipping_address_id = Column(Integer, ForeignKey("address.id"))

    billing_address = relationship("Address", foreign_keys=[billing_address_id])
    shipping_address = relationship("Address", foreign_keys=[shipping_address_id])

    def __repr__(self):
        return self.name,self.billing_address

class Address(Base):
    __tablename__ = 'address'
    id = Column(Integer, primary_key=True)
    street = Column(String(32))
    city = Column(String(32))
    state = Column(String(32))

    def __repr__(self):
        return self.street

Base.metadata.create_all(engine)   #创建表结构

session_class = sessionmaker(bind=engine)
session = session_class()

obj = session.query(Customer).filter(Customer.name=='克里斯').first()
print(obj.name,obj.billing_address,obj.shipping_address)

# addr1 = Address(street='浦沿',city='滨江',state='杭州')
# addr2 = Address(street='城站',city='江干',state='杭州')
# addr3 = Address(street='冰雪大世界',city='柯桥',state='绍兴')
# session.add_all([addr1,addr2,addr3])
#
#
# c1 = Customer(name='克里斯',billing_address=addr1,shipping_address=addr2)
# c2 = Customer(name='艾瑞克',billing_address=addr3,shipping_address=addr3)
# session.add_all([c1,c2])

session.commit()
多外键
  • 多对多关系

现在来设计一个能描述“图书”与“作者”的关系的表结构，需求是

  1. 一本书可以有好几个作者一起出版
  2. 一个作者可以写好几本书

表结构设计如下：

通过book_m2m_author表完成了book表和author表之前的多对多关联； 

#!/usr/bin/env python
# -*- coding:utf-8 -*-
#  Author fuyf

from sqlalchemy import Table, Column, Integer,String,DATE, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

book_m2m_author = Table('book_m2m_author', Base.metadata,
                        Column('book_id',Integer,ForeignKey('books.id')),
                        Column('author_id',Integer,ForeignKey('authors.id')),
                        )

class Book(Base):
    __tablename__ = 'books'
    id = Column(Integer,primary_key=True)
    name = Column(String(64))
    pub_date = Column(DATE)
    authors = relationship('Author',secondary=book_m2m_author,backref='books')

    def __repr__(self):
        return self.name

class Author(Base):
    __tablename__ = 'authors'
    id = Column(Integer, primary_key=True)
    name = Column(String(32))

    def __repr__(self):
        return self.name

engine = create_engine('mysql+pymysql://root:servyou@localhost/servyoudb?charset=utf8')
Base.metadata.create_all(engine)

session_class = sessionmaker(bind=engine)
session = session_class()

a1 = Author(name='克里斯')
a2 = Author(name='艾瑞克')
a3 = Author(name='杰克')

b1 = Book(name='跟chris学python')
b2 = Book(name='跟chris学linux')
b3 = Book(name='跟chris学吉他')

b1.authors = [a1,a2,a3]
b2.authors = [a1,a3]

session.add_all([a1,a2,a3,b1,b2,b3])
session.commit()

此时，手动连上mysql，分别查看这3张表，发现book_m2m_author中自动创建了多条纪录用来连接book和author表

mysql> select * from book_m2m_author order by book_id,author_id;
+---------+-----------+
| book_id | author_id |
+---------+-----------+
|       1 |         1 |
|       1 |         2 |
|       1 |         3 |
|       2 |         1 |
|       2 |         3 |
+---------+-----------+
5 rows in set (0.00 sec)

mysql> select * from authors;
+----+-----------+
| id | name      |
+----+-----------+
|  1 | 克里斯    |
|  2 | 艾瑞克    |
|  3 | 杰克      |
+----+-----------+
3 rows in set (0.00 sec)

mysql> select * from books;
+----+-------------------+----------+
| id | name              | pub_date |
+----+-------------------+----------+
|  1 | 跟chris学python   | NULL     |
|  2 | 跟chris学linux    | NULL     |
|  3 | 跟chris学吉他     | NULL     |
+----+-------------------+----------+
3 rows in set (0.00 sec)

mysql> select * from book_m2m_author order by book_id;
+---------+-----------+
| book_id | author_id |
+---------+-----------+
|       1 |         2 |
|       1 |         1 |
|       1 |         3 |
|       2 |         1 |
|       2 |         3 |
+---------+-----------+
5 rows in set (0.00 sec)

 查询：

#查询
print('--------通过书表查关联的作者---------')
book_obj = session.query(Book).filter(Book.name=='跟chris学python').first()
print(book_obj.name,book_obj.authors)

print('--------通过作者表查关联的书---------')
author_obj = session.query(Author).filter(Author.name=='克里斯').all()
print(author_obj[0].name,author_obj[0].books)

#输出
--------通过书表查关联的作者---------
跟chris学python [克里斯, 艾瑞克, 杰克]
--------通过作者表查关联的书---------
克里斯 [跟chris学python, 跟chris学linux]

多对多删除

删除数据时不用管boo_m2m_authors　， sqlalchemy会自动帮你把对应的数据删除

  •  通过书删除作者
author_obj = session.query(Author).filter_by(name="杰克").first()

book_obj = session.query(Book).filter_by(name="跟chris学linux").first()

book_obj.authors.remove(author_obj)  # 从一本书里删除一个作者
session.commit()
  • 直接删除作者　

删除作者时，会把这个作者跟所有书的关联数据也自动删除

author_obj =session.query(Author).filter_by(name="克里斯").first()
# print(author_obj.name , author_obj.books)
session.delete(author_obj)
session.commit()

处理中文

sqlalchemy设置编码字符集一定要在数据库访问的URL上增加charset=utf8，否则数据库的连接就不是utf8的编码格式

engine = create_engine('mysql+pymysql://root:servyou@localhost/servyoudb?charset=utf8',echo=True)

练习：学员管理系统

学员管理系统：
【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/fuyefeng/p/7138673.html