目录
- 一、redis之列表(List)类型
- 二、redis之hash(字典)类型
- 三、redis其他方法(所有类型通用的方法)
- 四、redis 管道
- 五、django中使用redis
- 六、celery介绍和安装
- 七、celery快速使用
- 八、celery包结构
- 九、作业
一、redis之列表(List)类型
List操作,redis中的List在在内存中按照一个name对应一个List来存储。如图:
lpush(name,values)
# 在name对应的list中添加元素,每个新的元素都添加到列表的最左边
# 如:
# r.lpush('oo', 11,22,33)
# 保存顺序为: 33,22,11
# 扩展:
# rpush(name, values) 表示从右向左操作
lpushx(name,value)
# 在name对应的list中添加元素,只有name已经存在时,值添加到列表的最左边
# 更多:
# rpushx(name, value) 表示从右向左操作
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,索引结束位置 print(re.lrange('aa',0,re.llen('aa')))
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),从右向左获取数据爬虫实现简单分布式:多个url放到列表里,往里不停放URL,程序循环取值,但是只能一台机器运行取值,可以把url放到redis中,多台机器从redis中取值,爬取数据,实现简单分布式
brpoplpush(src, dst, timeout=0)
# 从一个列表的右侧移除一个元素并将其添加到另一个列表的左侧
# 参数:
# src,取出并要移除元素的列表对应的name
# dst,要插入元素的列表对应的name
# timeout,当src对应的列表中没有数据时,阻塞等待其有数据的超时时间(秒),0 表示永远阻塞
自定义增量迭代
# 由于redis类库中没有提供对列表元素的增量迭代,如果想要循环name对应的列表的所有元素,那么就需要:
# 1、获取name对应的所有列表
# 2、循环列表
# 但是,如果列表非常大,那么就有可能在第一步时就将程序的内容撑爆,所有有必要自定义一个增量迭代的功能:
import redis
conn=redis.Redis(host='127.0.0.1',port=6379)
# conn.lpush('test',*[1,2,3,4,45,5,6,7,7,8,43,5,6,768,89,9,65,4,23,54,6757,8,68])
# conn.flushall()
def scan_list(name,count=2):
index=0
while True:
data_list=conn.lrange(name,index,count+index-1)
if not data_list:
return
index+=count
for item in data_list:
yield item
print(conn.lrange('test',0,100))
for item in scan_list('test',5):
print('---')
print(item)
ps:需要熟悉的方法
lpush
lpop
llen
lrange
操作代码
import redis
conn = redis.Redis()
# 1 lpush(name, values) 从左侧插入
# conn.lpush('girls', '刘亦菲', '迪丽热巴')
# conn.lpush('girls', '周淑怡')
# 2 rpush(name, values) 表示从右向左操作
# conn.rpush('girls', '小红')
# 3 lpushx(name, value)
# conn.lpushx('boys','小刚')
# conn.lpush('boys','小刚')
# conn.lpushx('girls','小刚')
# 4 rpushx(name, value) 表示从右向左操作
# 5 llen(name)
# res = conn.llen('girls')
# print(res)
# 6 linsert(name, where, refvalue, value))
# conn.linsert('girls','before','迪丽热巴','古力娜扎')
# conn.linsert('girls', 'after', '小红', '小绿')
# conn.linsert('girls', 'after', '小黑', '小嘿嘿') # 没有标杆,插入不进去
# 7 r.lset(name, index, value) # 按位置改值
# conn.lset('girls',1,'xxx')
# 8 r.lrem(name, value, num)
# conn.lrem('girls',1,'xxx') # 从左侧开始,删除1个
# conn.lrem('girls',-1,'xxx') # 从右侧开始,删除1个
# conn.lrem('girls',0,'xxx') # 从左开始,全删除
# 9 lpop(name)
# res=conn.lpop('girls')
# print(res)
# 10 rpop(name) 表示从右向左操作
# 11 lindex(name, index)
# res = str(conn.lindex('girls', 1), encoding='utf-8')
# print(res)
# 12 lrange(name, start, end)
# res=conn.lrange('girls',0,0) # 前闭后闭区间
# print(res)
# 13 ltrim(name, start, end)
# conn.ltrim('girls',2,3)
# 14 rpoplpush(src, dst)
# 15 blpop(keys, timeout) # 记住 ,可以做消息队列使用 阻塞式弹出,如果没有,就阻塞
# res=conn.blpop('boys')
# print(res)
# 16 r.brpop(keys, timeout),从右向左获取数据
# 17 brpoplpush(src, dst, timeout=0)
conn.close()
二、redis之hash(字典)类型
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时则创建(相当于添加)
# 批量添加操作
conn.hset('userinfo',mapping={'age':19,'hobby':'篮球'})
hmset(name, mapping)
ps:已经被弃用了,可以在hset中使用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的所有键值
print(re.hgetall('xxx').get(b'name'))
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的键值对删除
print(re.hdel('xxx','sex','name'))
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
ps:需要熟悉的方法
hset
hget
hmget
hlen
hdel
hscan_iter 获取所有值,但是省内存 等同于hgetall
操作代码
import redis
conn = redis.Redis()
# 1 hset(name, key, value)
# conn.hset('userinfo','name','lqz')
# conn.hset('userinfo',mapping={'age':19,'hobby':'篮球'})
# 2 hmset(name, mapping) # 批量设置,被弃用了,以后都使用hset
# conn.hmset('userinfo2',{'age':19,'hobby':'篮球'})
# 3 hget(name,key)
# res=conn.hget('userinfo','name')
# print(res)
# 4 hmget(name, keys, *args)
# res=conn.hmget('userinfo',['name','age'])
# res = conn.hmget('userinfo', 'name', 'age')
# print(res)
# 5 hgetall(name) # 慎用
# res=conn.hgetall('userinfo')
# print(res)
# 6 hlen(name)
# res=conn.hlen('userinfo')
# print(res)
# 7 hkeys(name)
# res=conn.hkeys('userinfo')
# print(res)
# 8 hvals(name)
# res=conn.hvals('userinfo')
# print(res)
# 9 hexists(name, key)
# res = conn.hexists('userinfo', 'name')
# res = conn.hexists('userinfo', 'name1')
# print(res)
# 10 hdel(name,*keys)
# res = conn.hdel('userinfo', 'age')
# print(res)
# 11 hincrby(name, key, amount=1)
conn.hincrby('userinfo', 'age', 2)
# article_count ={
# '1001':0,
# '1002':2,
# '3009':9
# }
# 12 hincrbyfloat(name, key, amount=1.0)
# hgetall 会一次性全取出,效率低,可以能占内存很多
# 分批获取,hash类型是无序
# 插入一批数据
# for i in range(1000):
# conn.hset('hash_test','id_%s'%i,'鸡蛋_%s号'%i)
# res=conn.hgetall('hash_test') # 可以,但是不好,一次性拿出,可能占很大内存
# print(res)
# 13 hscan(name, cursor=0, match=None, count=None) # 它不单独使用,拿的数据,不是特别准备
# res = conn.hscan('hash_test', cursor=0, count=5)
# print(len(res[1])) #(数字,拿出来的10条数据) 数字是下一个游标位置
# 咱们用这个,它内部用了hscan,等同于hgetall 所有数据都拿出来,count的作用是,生成器,每次拿count个个数
# 14 hscan_iter(name, match=None, count=None)
res=conn.hscan_iter('hash_test',count=10)
# print(res) # generator 只要函数中有yield关键字,这个函数执行的结果就是生成器 ,生成器就是迭代器,可以被for循环
# for i in res:
# print(i)
conn.close()
三、redis其他方法(所有类型通用的方法)
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
expire(name ,time)
# 为某个redis的某个name设置超时时间
rename(src, dst)
# 对redis的name(src)重命名为dst
move(name, db))
# 将redis的某个值移动到指定的db下
randomkey()
# 随机获取一个redis的name(key)(不删除)
type(name)
# 获取name对应值的类型
scan(cursor=0, match=None, count=None)
scan_iter(match=None, count=None)
# 同字符串操作,用于增量迭代获取key
import redis
conn = redis.Redis()
# 1 delete(*names)
# conn.delete('name', 'userinfo2')
# conn.delete(['name', 'userinfo2']) # 不能用它(可变长形参的知识点,列表不能被打散)
# conn.delete(*['name', 'userinfo2']) # 可以用它
# 2 exists(name)
# res=conn.exists('userinfo')
# print(res)
# 3 keys(pattern='*')
# res=conn.keys('w?e') # ?表示一个字符, * 表示多个字符
# print(res)
# 4 expire(name ,time)
# conn.expire('userinfo',3)
# 5 rename(src, dst)
# conn.rename('hobby','hobby111')
# 6 move(name, db))
# conn.move('hobby111',8)
# 7 randomkey()
# res=conn.randomkey()
# print(res)
# 8 type(name)
# print(conn.type('girls'))
print(conn.type('age'))
conn.close()
四、redis 管道
# 事务---》四大特性:
-原子性
-一致性
-隔离性
-持久性
# redis支持事务吗 单实例才支持所谓的事物,支持事务是基于管道的
-执行命令 一条一条执行
-张三 金额 -100 conn.decr('zhangsan_je',100)
中间代码挂了,就导致张三的钱少了,你的钱没多
-你 金额 100 conn.incr('李四_je',100)
- 把这两条命令,放到一个管道中,先不执行,执行excute,一次性都执行完成
conn.decr('zhangsan_je',100) conn.incr('李四_je',100)
pycharm使用redis事务
# 如何使用
import redis
conn = redis.Redis()
p=conn.pipeline(transaction=True)
p.multi()
p.decr('zhangsan_je', 100)
# raise Exception('崩了')
p.incr('lisi_je', 100)
p.execute()
conn.close()
五、django中使用redis
方式一
自定义包方案(通用的,不针对与框架,所有框架都可以用)
-第一步:写一个pool.py
import redis
POOL = redis.ConnectionPool(max_connections=100)
-第二步:以后在使用的地方,直接导入使用即可
import redis
from utils.pool import POOL
def test_redis(request):
con = redis.Redis(connection_pool=POOL)
con.incr('count')
res = con.get('count')
return JsonResponse({'count': '今天这个接口被访问次数为:%s' % res}, json_dumps_params={'ensure_ascii': False})
方式二
django 方案
在django中使用redis需要安装django-redis模块
pip install django-redis
方案一
django的缓存使用redis 【推荐使用】
这是只是用redis当作django的缓存,我们仍旧是使用cache操作
django-redis模块内部封装了pickle,我们在存取的时候会自动序列化和反序列化,因此我们在存数据的时候可以存对象进去
步骤一
在配置文件中添加配置(默认的配置文件中也有,在django的conf文件夹中找默认配置文件)
# redis配置
CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://127.0.0.1:6379",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {"max_connections": 100}
# "PASSWORD": "123",
}
}
}
views.py中测试
from django.core.cache import cache
# cache.set('count', 0)
'第一次运行的时候我们需要在redis中创建出count这个键值对'
def test_redis(request):
res = cache.get('count')
print(type(res))
print(res)
'我们会发现数据类型是整形'
cache.set('count', res + 1)
return JsonResponse({'count': '今天这个接口被访问次数为:%s' % str(res+1)}, json_dumps_params={'ensure_ascii': False})
结果如下
方案二
第三方:完全使用django-redis模块
因为这里是完全使用了django-redis模块,所以这里可以使用redis的数据类型,以及方法
from django_redis import get_redis_connection
def test_redis(request):
con = get_redis_connection()
con.incr('count')
print(con.get('count'))
return JsonResponse({'count': '今天这个接口被访问次数为:%s' % con.get('count')}, json_dumps_params={'ensure_ascii': False})
六、celery介绍和安装
官网
Celery 官网:http://www.celeryproject.org/
Celery 官方文档英文版:http://docs.celeryproject.org/en/latest/index.html
Celery 官方文档中文版:http://docs.jinkan.org/docs/celery/
Celery 是什么
- Celery 是异步任务框架,
- 翻译过来是 芹菜 的意思,但是跟芹菜没有关系(就像python跟蟒蛇没什么关系)
- 框架:运行后是一个服务,python的框架,跟django无关(两者运行互不影响)
Celery使用场景
异步执行:解决耗时任务
延迟执行:解决延迟任务
定时执行:解决周期(周期)任务
理解celery的运行原理
# 理解celery的运行原理
"""
1)可以不依赖任何服务器,通过自身命令,启动服务
2)celery服务为为其他项目服务提供异步解决任务需求的
注:会有两个服务同时运行,一个是项目服务,一个是celery服务,项目服务将需要异步处理的任务交给celery服务,celery就会在需要时异步完成项目的需求
人是一个独立运行的服务 | 医院也是一个独立运行的服务
正常情况下,人可以完成所有健康情况的动作,不需要医院的参与;但当人生病时,就会被医院接收,解决人生病问题
人生病的处理方案交给医院来解决,所有人不生病时,医院独立运行,人生病时,医院就来解决人生病的需求
"""
Celery架构
Celery的架构由三部分组成,消息中间件(message broker)、任务执行单元(worker)和 任务执行结果存储(task result store)组成。
任务(消息)中间件
Celery本身不提供消息服务,但是可以方便的和第三方提供的消息中间件集成。包括,RabbitMQ, Redis等等
ps:需要借助于第三方 redis rabbitmq
ps:真正执行异步任务的进程,celery提供的
任务执行单元
Worker是Celery提供的任务执行的单元,worker并发的运行在分布式的系统节点中。
任务结果存储
Task result store用来存储Worker执行的任务的结果,Celery支持以不同方式存储任务的结果,包括AMQP, redis等
ps:需要借助于第三方:redis,mysql
七、celery快速使用
celery 不支持win,需要通过eventlet支持在win上运行
安装celery与eventlet模块
pip install celery
pip install eventlet
快速使用
步骤一
创建main.py在内部编写celery任务中间件和任务结果存储的存储位置,并且创建celery对象,提供给外部调用,然后我们把需要执行的函数也定义在这里
from celery import Celery
broker = 'redis://127.0.0.1:6379/1'
'这里表示异步提交的需要等待执行的任务,存放在redis的编号为1的库中'
backend = 'redis://127.0.0.1:6379/2'
'这里是存放执行后的结果,存在redis的编号为2的库中'
app = Celery('test', broker=broker, backend=backend)
import time
@app.task
def add(a, b):
time.sleep(3)
print(a + b)
return a + b
步骤二
创建一个新的py文件handup.py,在这里编写代码提交任务
'提交任务'
from main import add
'同步调用'
# res = add(7,8)
# print(res)
'异步调用'
res = add.delay(6,6)
'在add后点delay方法就是异步调用,参数放在delay的括号内'
print(res)
'e9155a7f-d2fb-4bbc-87b2-ba992969a5c6'
'这里的res获取的到是一串字符,存放在redis的id为1的库中'
步骤三
启动worker(任务执行单元),执行任务
在执行worker命令之前,windows系统必须要安装eventlet模块
然后再执行命令前,我们需要去项目的main文件所在目录执行命令,因为这个命令的执行文件是main.py
启动worker命令
ps:-A后面跟的是celery对象所在的文件,-P是指定我们需要使用eventlet
win:
-4.x之前版本
celery worker -A main -l info -P eventlet
-4.x之后
celery -A main worker -l info -P eventlet
mac:
celery -A main worker -l info
步骤四
worker会执行消息中间件中的任务,把结果存到我们指定的redis的第二个库中
只要我们不主动关闭,celery worker就会一直运行
步骤五
查看执行结果,拿到执行的结果
上图是我们通过图形化界面看到的结果,正常来说我们需要用代码来获得运行的结果
这里我们创建一个新的py文件来获取结果,并且需要用到之前获取到的id来查询结果
get_result.py
from main import app
from celery.result import AsyncResult
id = '21325a40-9d32-44b5-a701-9a31cc3c74b5'
if __name__ == '__main__':
ans = AsyncResult(id=id, app=app)
if ans.successful():
result = ans.get()
print(result)
elif ans.failed():
print('任务失败')
elif ans.status == 'PENDING':
print('任务等待中被执行')
elif ans.status == 'RETRY':
print('任务异常后正在重试')
elif ans.status == 'STARTED':
print('任务已经开始被执行')
ps:将来我们在任意位置需要查看结果的时候,导入上述代码使用即可
八、celery包结构
包结构目录
project
├── celery_task # celery包
│ ├── __init__.py # 包文件
│ ├── celery.py # celery连接和配置相关文件,且名字必须交celery.py
│ └── tasks.py # 所有任务函数
├── add_task.py # 添加任务
└── get_result.py # 获取结果
步骤一
创建包celery_task,并且报下必须有一个叫celery.py的py文件
celery.py
from celery import Celery
# 提交的异步任务,放在里面
broker = 'redis://127.0.0.1:6379/1'
# 执行完的结果,放在这里
backend = 'redis://127.0.0.1:6379/2'
# 不要忘了include
'上面两个配置是设置存储数据的redis库位置'
'下面的app中多了一格include属性,他相当于是注册需要执行的函数'
app = Celery('test', broker=broker, backend=backend, include=['celery_task.order_task', 'celery_task.user_task'])
步骤二
接着我们我们在保内编写task文件,编写任务函数,我们可以根据用途分成不同的task文件
order_task
from .celery import app
import time
@app.task
def add(a, b):
print('-----', a + b)
time.sleep(2)
return a + b
user_task
from .celery import app
import time
@app.task
def send_sms(phone, code):
print("给%s发送短信成功,验证码为:%s" % (phone, code))
time.sleep(2)
return True
步骤三
提交任务到中间件,等待worker执行
handup.py
from celery_task.user_task import send_sms
from celery_task.order_task import add
'异步调用'
res = send_sms.delay(199999999, '88888')
print(res)
# res = add.delay(19, 3)
# print(res)
步骤四
包所在目录下,启动worker
celery -A celery_task worker -l info -P eventlet
步骤五
worker执行完,结果会被存到backend中
步骤六
使用代码查看结果
依旧是使用跟上面一样的代码,导入的东西需要改一下
get_result.py
from celery_task.celery import app
from celery.result import AsyncResult
id = '9a2070f7-34d0-421e-836e-f6c664967e3f'
if __name__ == '__main__':
ans = AsyncResult(id=id, app=app)
if ans.successful():
result = ans.get()
print(result)
elif ans.failed():
print('任务失败')
elif ans.status == 'PENDING':
print('任务等待中被执行')
elif ans.status == 'RETRY':
print('任务异常后正在重试')
elif ans.status == 'STARTED':
print('任务已经开始被执行')
九、作业
安装celery 使用celery的包结构
-写一个run.py
循环打印
1 异步计算add
2 发送短信
3 查看短信发送结果
4 查看add异步的结果
用户选了发送短信 ,输入手机号,就可以异步给这个手机号发送短信
短信已发送