开发笔记:Pyhton浅谈Python与中间件之Redis

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:catt1e
import time
import json
import redis
def get_data(db_type=\'tv\'):
\'\'\'获取数据
1. 从mysql获取
2. 从其他接口获取
\'\'\'
data = None
if db_type == \'tv\':
data = {\'ch\':[\'ch001\',\'ch002\'], \'mudan\':[\'牡丹A\',\'牡丹B\']}
elif db_type == \'pc\':
data = {\'lenovo\':[\'le-001\',\'le-002\'], \'dell\':[\'dell-1\',\'dell-2\']}
time.sleep(3)
return data
def show_data(data):
for k, v in data.items():
print(k, v)
def mid_cache(k, data, db_type="tv"):
\'\'\'缓存数据
\'\'\'
try:
client = None
if db_type == \'tv\':
client = redis.StrictRedis(host=\'localhost\', port=6379, db=0)
elif db_type == \'pc\':
client = redis.StrictRedis(host=\'localhost\', port=6379, db=1)
ret = client.set(k, json.dumps(data))
return ret
except Exception as e:
print(e)
return False
redis_type = {
"tv": 0,
"pc": 1
}

def get_cache(k, db_type="tv"):
\'\'\'从k取缓存数据\'\'\'
client = redis.StrictRedis(host=\'localhost\', port=6379, db=redis_type[db_type])
try:
data = client.get(k)
return json.loads(data)
except Exception as e:
print(e)
return False
def main():
k = \'tv\'
data = get_cache(k, db_type="pc")
if data:
print(\'from cache...\')
show_data(data)
else:
print(\'no data, 第一次取\')
data = get_data(db_type="pc")
mid_cache(k, data, db_type="pc")
if __name__ == \'__main__\':
main()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:catt1e
# redis_pub.py 发布端

import redis
client = redis.Redis()
channels = ["Jay", "SHE", \'Beyond\']
def main():
print("可以发布到任意一个频道：")
for i in channels:
print(i)
while True:
ch = input("输入频道：")
print("输入发送的信息(q：退出)：")
msg = input(\'>>\')
if msg == \'q\':
break
client.publish(ch, msg)
if __name__ == \'__main__\':
main()

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:catt1e
# redis_sub.py订阅端
import redis
client = redis.StrictRedis()
channels = ["Jay", "SHE", \'Beyond\']
# 得到pubsub对象
# 通过对象订阅频道
# listen监听频道
def subscribe():
s1 = client.pubsub()
s1.subscribe(channels)
s2 = client.pubsub()
s2.subscribe(\'SHE\')

s3 = client.pubsub()
s3.subscribe(channels[:2])
show_msg(s1, \'s1\')
show_msg(s2, \'s2\')
show_msg(s3, \'s3\')
def show_msg(sub_obj, sub_name):
for msg in sub_obj.listen():
if msg["type"] == "message":
print(f\'{sub_name} get {msg["data"].decode()} from {msg["channel"]}\')
def main():
subscribe()
if __name__ == \'__main__\':
main()