Python入门自学进阶-Web框架——38、redis、rabbitmq、git

缓存数据库redis：

NoSQL（Not only SQL）泛指非关系型的数据库。为了解决大规模数据集合多重数据类的挑战。

NoSQL数据库的四大分类：

• 键值（Key-Value）存储数据库
• 列存储数据库
• 文档型数据库
• 图形（Graph）数据库

redis是业界主流的key-value nosql数据库之一。redis主要用在linux类系统。

要在 Ubuntu 上安装 Redis，打开终端，然后输入以下命令：

$sudo apt-get update
$sudo apt-get install redis-server

在windows系统下，下载压缩包，解压缩后，就可以直接运行。

解压缩后的Redis-x64-5.0.14.1：

 在此目录下运行cmd，在命令窗口运行命令redis-server redis.windows.conf，启动redis服务器

 可以看到，服务的端口是6379。此启动方法表示临时服务安装成功。使用该指令创建的服务，不会再window service列表中出现redis服务名，此窗口关闭，临时服务会自动退出

后台服务安装启动指令：redis-server.exe --service-install redis.windows.conf --loglevel verbose（一定要先卸载已经安装的临时或者固定的服务，否则会出现错误）

卸载服务：redis-server --service-uninstall

启动服务指令：redis-server.exe --service-start

如果需要停止就执行 redis-server --service-stop // 停止服务

再打开一个cmd窗口，运行客户端：redis-cli.exe

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

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

 String操作：

使用：set 、get 、 keys *

 在使用set时，显示set语法：

 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(key,value) 、setex(key,value,time) 、psetex(key,time_ms,value) 、mset(*args,**kwargs) 、get(key) 、 mget(keys,*args)、 getset(key,value)、 getrange(key,start,end)、 setrange(key,offset,value)、setbit(key,offset,value) 、getbit(name,offset)、 bitcount(key,start=None,end=None)、 strlen(key)、 incr(self,key,amount=1)、 incrbyfloat(self,key,amount=1.0)、decr(self,key,amount=1) 、 append(key,value)

setbit/getbit/bitcount这一组命令来记录、统计用户登录信息。

Hash操作：

hset (key field value)、hget(key field)、 hmset(key,mapping)、 hmget(key field)、 hkeys(key)、 hgetall(key)、 hlen(key)、 hvals(key)、 hexists(key field)、 hincrby(key field)、  HSCAN key cursor [MATCH pattern] [COUNT count]、 hdel(key *field)、 

list操作：

lpush、 llen、 lrange、 pushx、 lpushx、 linsert、lset、 lrem、 lpop、 lindex、 lrange、 ltrim、 rpoplpush、 blpop、 brpoplpush

set操作：

无序集合

sadd、scard 、 sdiff、 sdiffstore、 sinter、 sinterstore、 sismember、 smove、 spop、 srandmember、 srem、 sunion、 sunionstore、 sscan

有序集合

zadd、 zcard、 zrank、 zrem、 zrenrangebyrank、 zcount、 zincrby、 zrange、 zremrangebyscore、 zscore、 zinterstore、 zscan、

其他操作：

delete、exists、 keys 、 expire 、rename、 move、 randomkey、 type、 scan

python连接使用redis：

安装模块：pip install redis

使用：

import redis

r = redis.Redis(host='127.0.0.1',port=6379)
r.set('foo','bar')
print(r.get('foo'))

将host地址改为网卡的地址：

import redis

r = redis.Redis(host='192.168.1.117',port=6379)
r.set('foo','bar')
print(r.get('foo'))

结果是：ConnectionRefusedError: [WinError 10061] 由于目标计算机积极拒绝，无法连接。

需要修改redis.windows.conf文件中，将bind 127.0.0.1 改为bind 192.168.1.117

如果改为bind 0.0.0.0 则使用192.168.1.117或127.0.0.1都能访问到。

这时，是未经验证就直接访问了，可以使用密码进行验证。修改redis.windows.conf文件中的配置项：# requirepass foobared

可以看到，需要密码是注释掉的，即不需要密码验证，打开验证：requirepass abc123

重启后，再次运行上面的程序：提示：redis.exceptions.AuthenticationError: Authentication required.

现在需要验证了

import redis

r = redis.Redis(host='127.0.0.1',port=6379,password='abc123')
r.set('foo','bar')
print(r.get('foo'))

此时，连接成功。

在客户端使用时，也需要密码，还用auth password验证。

 使用连接池：

import redis

pool =redis.ConnectionPool(host='192.168.1.117',port=6379,password='abc123')
r = redis.Redis(connection_pool=pool)
r.set('foo2','bar222')
print(r.get('foo2'))
r.set('ccc2',1234)

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

import redis

pool =redis.ConnectionPool(host='192.168.1.117',port=6379,password='abc123')
r = redis.Redis(connection_pool=pool)

pipe = r.pipeline(transaction=True)
pipe.set('name','aaaaaaaa')
pipe.set('role','bbbbbbbbbbbb')
pipe.execute()

消息队列RabbitMQ：

消息队列中间件——是分布式系统中重要的组件，主要解决应用耦合，异步消息，流量削锋等问题实现高性能，高可用，可伸缩和最终一致性。 使用较多的消息队列有 ActiveMQ（安全），RabbitMQ，ZeroMQ，Kafka（大数据），MetaMQ，RocketMQ
RabbitMQ ——一个由 Erlang 语言开发的 AMQP 的开源实现。
AMQP ：Advanced Message Queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。
RabbitMQ 最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。具体特点包括：
1.可靠性（Reliability）
RabbitMQ 使用一些机制来保证可靠性，如持久化、传输确认、发布确认。
2.灵活的路由（Flexible Routing）
在消息进入队列之前，通过 Exchange 来路由消息的。对于典型的路由功能，RabbitMQ已经提供了一些内置的 Exchange 来实现。针对更复杂的路由功能，可以将多个Exchange 绑定在一起，也通过插件机制实现自己的 Exchange 。
3.消息集群（Clustering）
多个 RabbitMQ 服务器可以组成一个集群，形成一个逻辑 Broker
4.高可用（Highly Available Queues）
队列可以在集群中的机器上进行镜像，使得在部分节点出问题的情况下队列仍然可用。
5.多种协议（Multi-protocol）
RabbitMQ 支持多种消息队列协议，比如 STOMP、MQTT 等等。
6.多语言客户端（Many Clients）
RabbitMQ 几乎支持所有常用语言，比如 Java、.NET、Ruby 等等。
7.管理界面（Management UI）
RabbitMQ 提供了一个易用的用户界面，使得用户可以监控和管理消息 Broker 的许多方面。
8.跟踪机制（Tracing）
如果消息异常，RabbitMQ 提供了消息跟踪机制，使用者可以找出发生了什么。
9.插件机制（Plugin System）
RabbitMQ 提供了许多插件，来从多方面进行扩展，也可以编写自己的插件。

RabbitMQ的工作原理：

 

Broker：接收和分发消息的应用，RabbitMQ Server就是Message Broker
Connection： publisher / consumer（发布者/消费者）和 broker之间的TCP连接。（个人感觉用producer/consumer——生产者/消费者或publisher/subscriber——发布者/订阅者更合适）        Channel：如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection 内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id 帮助客户端和message broker识别 channel，所以channel 之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建TCP connection的开销
Exchange：message 到达 broker 的第一站，根据分发规则，匹配查询表中的 routing key，分发消息到queue 中去。常用的类型有: direct (point-to-point), topic(publish-subscribe) and fanout
(multicast)
Routing Key：生产者将消息发送到交换机时会携带一个key,来指定路由规则
binding Key：在绑定Exchange和Queue时，会指定一个BindingKey,生产者发送消息携带的RoutingKey会和bindingKey对比，若一致就将消息分发至这个队列
vHost 虚拟主机：每一个RabbitMQ服务器可以开设多个虚拟主机，每一个vhost本质上是一个mini版的RabbitMQ服务器，拥有自己的 "交换机exchange、绑定Binding、队列Queue"，更重要的是每一个vhost拥有独立的权限机制，这样就能安全地使用一个RabbitMQ服务器来服务多个应用程序，其中每个vhost服务一个应用程序。

RabbitMQ的安装（windows版本）：

RabbitMQ需要Erlang的支持，先安装这个支持包otp_win64_25.3.2.exe，OTP代表开放电信平台。 它是一个应用程序操作系统和一组用于构建大规模，容错，分布式应用程序的库和过程。核心概念是OTP行为，可以看作一个用回调函数作为参数的应用程序框架。

 

 

 

 

 

 

 

 安装RabbitMQ：rabbitmq-server-3.12.1.exe

 

 

 

安装完成后，在服务列表中能可见：

 

 安装管理界面（插件）：

进入rabbitMQ安装目录的sbin目录，点击上方的路径框输入cmd，按下回车键

 

 

运行命令：  rabbitmq-plugins enable rabbitmq_management 

打开浏览器，输入http://127.0.0.1:15672/，登录

 

输入用户名和密码，初始都为guest

 

 至此，安装完毕。

Python使用RabbitMQ，需要安装pika模块：pip install pika

 RabbitMQ的使用：最简单的发布与接收

# File：sender.py 发送消息，即生产者
import pika  # 链接mq需要pika模块

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

# 声明queue
channel.queue_declare(queue='hello')

# 在RabbitMQ中一个消息不可能被直接发送到queue，即队列中，它总是需要通过exchange进行转发
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='消息体：hello world!'.encode('utf-8'),)
print("[x] sent 'hello word!'")
connection.close()

 

# File：recv.py  消费者，即接收者、订阅者
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    'localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print("[x] Received %r" % body.decode("utf-8"))

channel.basic_consume(queue='hello',on_message_callback=callback,auto_ack=True)

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

在消费者端，basic_consume方法中设置了auto_ack=True，即自动应答，即消费者接受消息后，自动应答RabbitMQ服务器，即Broker，然后在Broker中的队列中将此消息删除，否则，如果设置为False，则会在队列中一直存在，如下：

 显示有2条未应答，消息还有2条。

可以设置手动应答：

# File：recv.py  消费者，即接收者、订阅者
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(
    'localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print("[x] Received %r" % body.decode("utf-8"))
    # 如果basic_consume中auto_ack为False，则这里要手动进行应答
    channel.basic_ack(delivery_tag=method.delivery_tag) # 手动应答
    print('手动应答队列中消息')
channel.basic_consume(queue='hello',on_message_callback=callback,auto_ack=False)

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

 自带的用户、密码为guest的只能本机登录，可以增加用户：

rabbitmqctl add_user 用户名 密码

给指定用户添加管理员权限：

rabbitmqctl  set_user_tags 用户名 administrator

给用户添加权限

rabbitmqctl set_permissions -p / 用户名 ".*" ".*" ".*"

以上是使用命令进行用户的添加，但是在测试时，出现问题：

 查询网上的解决方法，也没成功，然后从web端增加了用户。

使用增加的用户进行消息操作：

生产者：

# File：sender.py 发送消息，即生产者
import pika  # 链接mq需要pika模块
import time
user_info = pika.PlainCredentials('tester','test1234')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.1.117',5672,'/',user_info))
channel = connection.channel()

# 声明queue
# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句
channel.queue_declare(queue='hello')

# 在RabbitMQ中一个消息不可能被直接发送到queue，即队列中，它总是需要通过exchange进行转发
for i in range(0,20):
    channel.basic_publish(exchange='', # 简单模式，这里设置为空字符串就可以
                          routing_key='hello', # 指定消息要发送到哪个queue
                          body='消息体{}：hello world!'.format(i).encode('utf-8'), ) # 指定要发送的消息
    print("[x] sent %s 'hello word!'" % i)
    time.sleep(1)
connection.close()

#RabbitMQ中所有的消息都要先通过交换机，空字符串表示使用默认的交换机

消费者：

# File：recv.py  消费者，即接收者、订阅者
import pika

user_info = pika.PlainCredentials('tester','test1234')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.1.117',5672,'/',user_info))
channel = connection.channel()
channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):
    print("[x] Received %r" % body.decode("utf-8"))
    # 如果basic_consume中auto_ack为False，则这里要手动进行应答
    channel.basic_ack(delivery_tag=method.delivery_tag) # 手动应答
    print('手动应答队列中消息')
    # ch,即channel: 包含channel的一切属性和方法
    # method: 包含 consumer_tag, delivery_tag, exchange, redelivered, routing_key
    # properties: basic_publish 通过 properties 传入的参数
    # body: basic_publish发送的消息


channel.basic_consume(queue='hello', # 接收指定queue的消息
                      on_message_callback=callback, # 设置收到消息的回调函数
                      auto_ack=False) # 指定为True，表示消息接收到后自动给消息发送方回复确认，已收到消息，False表示不自动确认，需要在callback中手工确认

print('[*] Waiting for message. To exit press CTRL+C')

# 一直处于等待接收消息的状态，如果没收到消息就一直处于阻塞状态，收到消息就调用上面的回调函数
channel.start_consuming()

运行结果：

 以上是RabbitMQ简单模式

上面的生产者存在一个问题，就是在我们的消费者还没开始消费完队列里的消息，如果这时rabbitmq服务挂了，那么消息队列里的消息将会全部丢失，解决方法是在声明队列时，声明队列为可持久化存储队列，并且在生产者在将消息插入到消息队列时，设置消息持久化存储，具体如下

# File：sender.py 发送消息，即生产者
import pika  # 链接mq需要pika模块
import time
user_info = pika.PlainCredentials('tester','test1234')
connection = pika.BlockingConnection(pika.ConnectionParameters(
    '192.168.1.117',5672,'/',user_info))
channel = connection.channel()

# 声明queue
# 如果指定的queue不存在，则会创建一个queue，如果已经存在 则不会做其他动作，官方推荐，每次使用时都可以加上这句
channel.queue_declare(queue='durable_queue',durable=True)
#PS：这里不同种队列不允许名字相同,这个队列设置了持久化为True，即是一个持久化队列

# 在RabbitMQ中一个消息不可能被直接发送到queue，即队列中，它总是需要通过exchange进行转发
for i in range(0,20):
    channel.basic_publish(exchange='', # 简单模式，这里设置为空字符串就可以
                          routing_key='durable_queue', # 指定消息要发送到哪个queue
                          body='消息体{}：hello world!'.format(i).encode('utf-8'), # 指定要发送的消息
                          properties=pika.BasicProperties(delivery_mode=2))
                         # 设置当前消息持久化存储（properties=pika.BasicProperties(delivery_mode=2)）
    print("[x] sent %s 'hello word!'" % i)
    time.sleep(1)
connection.close()

#RabbitMQ中所有的消息都要先通过交换机，空字符串表示使用默认的交换机

消费者，只需修改队列的声明与生产者一致即可，如channel.queue_declare(queue='durable_queue',durable=True)

同时开启多个消费者，会随机读取队列中的消息。

默认安装的RabbitMQ中，在运行了上面的程序后，在RabbitMQ中connections、channel、exchange、queues信息：