目录
- 一、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，-l后面的info是用于指定打印信息的级别（类似异常处理的提示）
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异步的结果
用户选了发送短信 ，输入手机号，就可以异步给这个手机号发送短信
短信已发送