0


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信息:

标签: redis 数据库 缓存

本文转载自: https://blog.csdn.net/kaoa000/article/details/131687017
版权归原作者 kaoa000 所有, 如有侵权,请联系我们删除。

“Python入门自学进阶-Web框架——38、redis、rabbitmq、git”的评论:

还没有评论