'''
pip install redis
redis 可以看成内存中的大字典
redis五大数据类型 --> 指的是第一层的value值的类型
- 字符串 "k1"
- 列表 [11,22,33，11]
- 集合 {11,22,33}
- 字典 {
'kk1':vv1,
'kk2':vv2,
'kk3':{} 这个value不能是字典，如果有需要，则需要转换成字符串的形式。
}
- 有序集合 {('alex',4),('standby',3),('egon',9)}
'''
import redis
conn = redis.Redis(host='10.0.0.2',port=6379)
# 字符串
conn.set('k1','v1',ex=10)
conn.get('k1')
# 列表
# 可以当消息队列来使用
conn.lpush('users','alex')
conn.rpush('users','eric')
conn.lpop('users')
conn.rpop('users')
# 双向队列，可以hang住，可以模拟栈和队列， 爬虫，爬网页-广度优先和深度优先
conn.blpop('users',timeout=10)
conn.brpop('users',timeout=10)
conn.expire('key','超时时间')
count = conn.llen('users')
print(count)
vals = conn.lrange('users',0,10)
print(vals)
# redis针对列表类型，没有提供 iter 方法，需要自己实现
def list_scan_iter(key):
# count = conn.llen(key)
start = 0
step = 3
while True:
vals = conn.lrange(key,start,start+step)
start = start + step + 1
if not vals:
return
for val in vals:
yield val
result = list_scan_iter('users')
for item in result:
print(item)
# 集合
# 哈希
# 有序结合
遍历DB
import redis
conn = redis.StrictRedis(host='8.8.8.8', port=6379, db=8, password='1234567890')
cursor_num, keys = conn.execute_command('scan', 0, "count", 100)
while True:
print keys
if 0==int(cursor_num):
break
cursor_num, keys = conn.execute_command('scan', cursor_num, "count", 100)
Django通过redis实现session共享示例
安装插件 Django 1.11 以上版本才可以使用
pip2.7 install django-redis
配置插件
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'default'
CACHES = {
"default": {
"BACKEND": "django_redis.cache.RedisCache",
"LOCATION": "redis://10.10.10.10:6379",
"OPTIONS": {
"CLIENT_CLASS": "django_redis.client.DefaultClient",
"CONNECTION_POOL_KWARGS": {"max_connections": 100},
"PASSWORD": "your_password_for_redis",
}
}
}
使用redis
from django_redis import get_redis_connection
conn = get_redis_connection("default")
使用redis存储session信息
class UserAuth(object):
def __init__(self, request, response=None, user=None, autologin=None):
self.request = request
self.response = response
if user:
self.uid = str(user.id)
self.name = str(user.username)
self.autologin = autologin
self.sessionkey = None
self.logger = get_logger('UserAuth')
def session_check(self):
self.sessionkey = self.request.COOKIES.get('yoursessionkey', None)
if self.sessionkey and conn.exists('yoursessionkey_%s' % self.sessionkey):
return conn.hget('yoursessionkey_%s' % self.sessionkey, 'userinfo')
return None
def set_cookie_session(self):
self.sessionkey = self.request.COOKIES.get('yoursessionkey', None)
ttl = 60*60*24 if self.autologin else 60*60
if not self.sessionkey:
# set cookie
h = hashlib.md5()
h.update(str(int(time.time())))
self.sessionkey = h.hexdigest()
self.response.set_cookie('yoursessionkey', h.hexdigest(), max_age=ttl)
self.logger.error('New sessionid: {} for {}'.format(self.sessionkey,self.name))
key = 'yoursessionkey_%s' % self.sessionkey
if not conn.exists(key):
# set session
conn.hset(key, 'userinfo', '_'.join([self.name,self.uid]))
conn.expire(key,ttl)
self.logger.error('New redis key: {}, ttl: {}'.format(key,ttl))
return True
登录的视图函数
def verify(raw_password,encoded):
# 以PBKDF2PasswordHasher这个算法为例
from django.contrib.auth.hashers import PBKDF2PasswordHasher
hasher = PBKDF2PasswordHasher()
algorithm, iterations, salt, hash = encoded.split('$', 3)
encoded2 = hasher.encode(raw_password, salt, int(iterations))
return encoded == encoded2
class LoginForm(forms.Form):
username = forms.CharField()
password = forms.CharField(widget=forms.PasswordInput)
autologin = forms.BooleanField(required=False)
next = forms.CharField(widget=forms.HiddenInput)
def user_login(request):
next = request.GET.get('next',reverse('index'))
if 'POST' == request.method:
form = LoginForm(request.POST)
if form.is_valid():
name = form.cleaned_data['username']
pwd = form.cleaned_data['password']
autologin = form.cleaned_data['autologin']
user = UserInfo.objects.filter(username=name)
if user[0] and verify(pwd,user[0].password):
response = redirect(next)
user_auth = UserAuth(request, response, user[0], autologin)
if user_auth.set_cookie_session():
user.update(last_login=time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
return response
context = {'form':form,'errors':'用户名或密码错误'}
return render(request,'login.html',context)
else:
context = {'form':form,'errors':'用户名或密码不能为空'}
return render(request,'login.html',context)
form = LoginForm({'next':next})
context = {'form':form}
return render(request,'login.html',context)
登录验证中间件
import re
from django.shortcuts import redirect
from django.utils.deprecation import MiddlewareMixin
from django.conf import settings
class UserAuthMiddleware(MiddlewareMixin):
def process_request(self,request):
# Set white list which not check login
current_request_url = request.path_info
for url in settings.VALID_URL_LIST:
if re.match(url, current_request_url):
return None
user_auth = UserAuth(request)
if user_auth.session_check():
return None
else:
import urllib
params = urllib.urlencode({'next':request.path_info})
return redirect('/userlogin' + "?" + params)