第五章

目录

1. 消息队列
2. redis

---

1.消息队列

RabbitMQ安装

       http://www.cnblogs.com/shanyou/p/4067250.html

• 安装erlang http://www.erlang.org/
• 安装rabbitmq http://www.rabbitmq.com/download.html
• 安装 pika https://pika.readthedocs.io/en/0.10.0/

#!/usr/bin/env python
import pika
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()
channel.queue_declare(queue='hello')
channel.basic_publish(exchange='',
                      routing_key='hello',
                      body='Hello World'
                      )
print("[x] sent 'Hello World!'")
connection.close()

#!/usr/bin/env python
__author__ = 'han'
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)
channel.basic_consume(callback,
                      queue='hello',
                      )   #no_ack=True 不和服务端确认 
print('[*]Waiting for messages.to exit press CTRL+C')
channel.start_consuming()

• 持久化

import pika
#建立连接
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))  #建立连接
#建立管道
channel = connection.channel()   
#建立消息队列，durable=True 开启持久化
channel.queue_declare(queue='hello1',durable=True)   
channel.basic_publish(exchange='',
                      routing_key='hello1',   #指定消息队列
                      body='Hello World',     #发送消息
                      properties=pika.BasicProperties(   #消息持久化
                          delivery_mode=2, 
                      )
                      )
print("[x] sent 'Hello World!'")
connection.close()

import pika
#建立连接
connection = pika.BlockingConnection(pika.ConnectionParameters('127.0.0.1'))
#建立管道
channel = connection.channel()
#建立消息队列，durable=True开启持久化
channel.queue_declare(queue='hello1',durable=True)
def callback(ch,method,properties,body):
    print("[x]Received %r"% body)
    ch.basic_ack(delivery_tag=method.delivery_tag)  #收到消息回复生产者
channel.basic_qos(prefetch_count=1)   #负载均衡(处理完一个消息在发送下一个)
channel.basic_consume(callback,
                      queue='hello1',
                      )   #no_ack=True  不回复生产者
print('[*]Waiting for messages.to exit press CTRL+C')
channel.start_consuming()

查看消息持久化

cmd 
D:Program FilesRabbitMQ Server
abbitmq_server-3.6.11sbin>rabbitmqctl.bat list_queues 
Listing queues

• 广播

1. fanout: 所有bind到此exchange的queue都可以接收消息
2. direct: 通过routingKey和exchange决定的那个唯一的queue可以接收消息
3. topic:所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息

fanout

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
                         exchange_type="fanout")  #指定广播模式，不同版本可能是type="fanout"
message = ' '.join(sys.argv[1:]) or "info:Hello World!"  #发送的内容 argv1 或者 "info:Hello World!"
channel.basic_publish(exchange='logs',
                      routing_key='', 
                      body=message)
print("[x] Sent %r" % message)
connection.close() 

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
import pika
__author__ ='han'
connection = pika.BlockingConnection(pika.ConnectionParameters(host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='logs',
                         exchange_type='fanout')   #指定广播模式，不同版本可能是type="fanout"
result = channel.queue_declare(exclusive=True) #不指定queue名字,rabbit会随机分配一个名字,exclusive=True会在使用此queue的消费者断开后,自动将queue删除
queue_name = result.method.queue  #消息队列的名称
channel.queue_bind(exchange='logs',
                   queue=queue_name)    #制定消息队列的名称
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch, method, properties,body):
    print(" [x] %r" % body)
channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)
channel.start_consuming() 

direct

#接收指定的内容

#!/usr/bin/ebv python
# _*_ coding:utf-8 _*_
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(host="127.0.0.1"))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',   #设置direct模式
                         exchange_type='direct')
severity = sys.argv[1] if len(sys.argv) > 1 else 'info' #设置默认info
message = ' '.join(sys.argv[2:]) or 'Hello World!'
channel.basic_publish(exchange='direct_logs',   #调用direct模式
                      routing_key=severity,      #调用severity
                      body=message)
print(" [x] Sent %r:%r" % (severity,message))
connection.close()

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
      host='127.0.0.1'))
channel = connection.channel()
channel.exchange_declare(exchange='direct_logs',  #设置direct模式
                         exchange_type='direct')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
severities = sys.argv[1:]
if not severities:
    sys.stderr.write("usege: %s [info] [warning][reeor]
"%sys.argv[0])
    sys.exit(1)
for severity in severities:
    channel.queue_bind(exchange='direct_logs', #调用direct模式
                       queue=queue_name,
                       routing_key=severity)  #调用severities
print(' [*] Waiting for logs. To exit press CTRL+C')
def callback(ch,method,properties,body):
    print(" [x] %r:%r" % (method.routing_key, body))
channel.basic_consume(callback,
                      queue=queue_name,
                      no_ack=True)
channel.start_consuming()

topic

#过滤模式

 import pika
  import sys
  connection = pika.BlockingConnection(pika.ConnectionParameters(
      host='localhost'))
  channel = connection.channel()
  channel.exchange_declare(exchange='topic_logs',
                           exchange_type='topic')
  routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
  message = ' '.join(sys.argv[2:]) or 'Hello World!'
  channel.basic_publish(exchange='topic_logs',
                        routing_key=routing_key,
                        body=message)
  print(" [x] Sent %r:%r" % (routing_key, message))
  connection.close()

  import pika
  import sys
  connection = pika.BlockingConnection(pika.ConnectionParameters(
      host='localhost'))
  channel = connection.channel()
  channel.exchange_declare(exchange='topic_logs',
                           exchange_type='topic')
  result = channel.queue_declare(exclusive=True)
  queue_name = result.method.queue
  binding_keys = sys.argv[1:]
  if not binding_keys:
      sys.stderr.write("Usage: %s [binding_key]...
" % sys.argv[0])
      sys.exit(1)
  for binding_key in binding_keys:
      channel.queue_bind(exchange='topic_logs',
                         queue=queue_name,
                         routing_key=binding_key)
  print(' [*] Waiting for logs. To exit press CTRL+C')
  def callback(ch, method, properties, body):
      print(" [x] %r:%r" % (method.routing_key, body))
  channel.basic_consume(callback,
                        queue=queue_name,
                        no_ack=True)
  channel.start_consuming()

  1. To receive all the logs run:
  python receive_logs_topic.py "#"

  2. To receive all logs from the facility "kern":
  python receive_logs_topic.py "kern.*"

  3. Or if you want to hear only about "critical" logs:
  python receive_logs_topic.py "*.critical"

  4. You can create multiple bindings:
  python receive_logs_topic.py "kern.*" "*.critical"

  5. And to emit a log with a routing key "kern.critical" type:
  python emit_log_topic.py "kern.critical" "A critical kernel error" 

rpc

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
import pika
import sys
connection = pika.BlockingConnection(pika.ConnectionParameters(
                                     host='127.0.0.1'))
channel = connection.channel()
channel.queue_declare(queue='rpc_queue') #设置rpc模式
#斐波那契
def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n-1) + fib(n-2)
#2.
def on_request(ch,method,props,body):
    n = int(body) #斐波那契
    print("[.]fib(%s)"%n)
    response = fib(n)
    #3.发送
    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,  #接收到的随机queue
                     properties=pika.BasicProperties(correlation_id=    #接收到的uuid
                                                     props.correlation_id),
                     body = str(response)) #发送消息
    ch.basic_ack(delivery_tag = method.delivery_tag)  #等待确认
#1.开始接收
channel.basic_consume(on_request,queue='rpc_queue')
print("[x] Awaiting RPC requests")
channel.start_consuming()

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
import pika
import sys
import uuid
class FibonacCiRpcClient(object):
    def __init__(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(
                                                  host = '127.0.0.1'))
        self.channel= self.connection.channel()
        result = self.channel.queue_declare(exclusive=True)   #设置随机queue
        self.callback_queue = result.method.queue
        self.channel.basic_consume(self.on_response,no_ack=True,   #设置接收
                                   queue=self.callback_queue)
    def on_response(self,ch,method,props,body):
        if self.corr_id == props.correlation_id:   #确认server和clinet的uuid是否相等
            self.response = body
    #1.发送
    def call(self,n):
        self.response = None
        self.corr_id = str(uuid.uuid4())
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue',     #设置rpc模式
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue,  #指定随机queue
                                          correlation_id= self.corr_id), #uuid
                                   body=str(n))                          #发送消息
        #2.没有数据开始接收
        while self.response is None:
            self.connection.process_data_events()  #设置非堵塞模式
        return int(self.response)                 #返回结果
fibonacci_rpc =  FibonacCiRpcClient()
print(" [x] Requesting fib(5)")
response = fibonacci_rpc.call(5)
print('[.]Got%r'%response)

2.redis

http://www.cnblogs.com/wupeiqi/articles/5132791.html

redis安装

yum install epel-release 
yum install redis 
redis-server /etc/redis.conf
redis-cli -h 192.168.80.11

• python执行rerdis命令

redis-py提供两个类Redis和StrictRedis用于实现Redis的命令，StrictRedis用于实现大部分官方的命令，并使用官方的语法和命令，Redis是StrictRedis的子类，用于向后兼容旧版本的redis-py。

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
__author__ = 'han'
import redis
r = redis.Redis(host='192.168.80.12',port=6379)
r.set('foo','Bar')  #设置一个字典
print(r.get('foo'))  

• 链接池

redis-py使用connection pool来管理对一个redis server的所有连接，避免每次建立、释放连接的开销。默认，每个Redis实例都会维护一个自己的连接池。可以直接建立一个连接池，然后作为参数Redis，这样就可以实现多个Redis实例共享一个连接池。

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
__author__ = 'han'
import redis
pool = redis.ConnectionPool(host='192.168.80.12',port=6379)
r = redis.Redis(connection_pool=pool)
r.set('foo','bar')
print(r.get('foo'))

• 字典

查看修改等..

set name han  #设置一个字典

mset naem8 zhang name9 li  #设置多个 


192.168.80.12:6379> keys *  #查看所有 
1) "naem" 
2) "aeg" 
3) "name" 
4) "foo" 


set test2 100 ex 5  #设置5秒  

setex name4 3 alex  #设置alex3秒


set 2b alex nx         #不存在设置

set 2b abc xx   #存在设置 

getset name9 jack  #设置新值并获取原来的值

getrange name9 0 2 #切片 

setrange name9 0 l  #修改

　

1.set naem11 alex
 ord('a')
97

 bin(97)
'0b1100001'

   127 64 32 16 8 4 2 1 
   0  97 1  1  0  0 0 0 1

 192.168.80.12:6379> setbit naem11 6 1
(integer) 0

192.168.80.12:6379> get naem11
"clex"

setbit

模拟用户在线
192.168.80.12:6379> setbit n5 1000 1
(integer) 0
192.168.80.12:6379> setbit n5 55 1
(integer) 0
192.168.80.12:6379> setbit n5 600 1
(integer) 0
192.168.80.12:6379> bitcount n5
(integer) 3                         共3 
192.168.80.12:6379> getbit n5 55  查看
(integer) 1

incrbyfloat

#支持小数

192.168.80.12:6379> set n6 1.0    
OK
192.168.80.12:6379> incrbyfloat n6 1.2
"2.2"
192.168.80.12:6379> incrbyfloat n6 1.4
"3.6"

incr

模拟用户在线
192.168.80.12:6379> incr name8 
(integer) 1
192.168.80.12:6379> incr name8 
(integer) 2
192.168.80.12:6379> incr name8  加
(integer) 3
192.168.80.12:6379> decr name8  减
(integer) 2
192.168.80.12:6379> decr name8 
(integer) 1

append

#追加

192.168.80.12:6379> set name9 zhang 设置字典
OK
192.168.80.12:6379> append name9 san   追加为zhangsan
(integer) 8
192.168.80.12:6379> get name9
"zhangsan"

• Hash

#设置
127.0.0.1:6379> hset ni name li 
(integer) 1

#长度
127.0.0.1:6379> hlen ni   (integer) 1
192.168.80.12:6379> hset info id 999  
(integer) 1

#设置
192.168.80.12:6379> hset info name alex
(integer) 1
192.168.80.12:6379> hget info name
"alex"

#查看keys
192.168.80.12:6379> hkeys info 
1) "id"
2) "name"

 查看values
192.168.80.12:6379> hvals info
1) "999"
2) "alex"

hmset

#设置多个
192.168.80.12:6379> hmset info2 k1 1 k2 2  
OK

#查看keys
192.168.80.12:6379> keys info2  
1) "info2"

#查看所有
192.168.80.12:6379> hgetall info2 
1) "k1"
2) "1"

hmget

#查看多个
1) "1"
192.168.80.12:6379> hmget info2 k1 k2 
2) "2"

hexists

#判断一个keys是否存在
192.168.80.12:6379> hexists info2 k3 
(integer) 0

192.168.80.12:6379> hexists info2 k2
(integer) 1

hscan

#过滤查看
192.168.80.12:6379> hscan info2 0 match k* 
1) "0"
2) 1) "k1"
   2) "1"
   3) "k2"
   4) "2"

• 列表

设置列表

192.168.80.12:6379> lpush names alex hanyan zhuangyang
(integer) 3

#从右查看
192.168.80.12:6379> lrange names 0 -1  
1) "zhuangyang"
2) "hanyan"
3) "alex"


192.168.80.12:6379> rpush names wangsen zouyi
(integer) 5

#从左查看
192.168.80.12:6379> lrange names 0 -1
1) "zhuangyang"
2) "hanyan"
3) "alex"
4) "wangsen"
5) "zouyi"

#查看长度
127.0.0.1:6379> lpush n1 zhang
(integer) 1
127.0.0.1:6379> llen n1
(integer) 1

lrem

#删除
127.0.0.1:6379> lrem n1 1 zhang  
(integer) 1

#从左删除
127.0.0.1:6379> lpop n1   
"han"

#从右删除
127.0.0.1:6379> rpop n1  
"zhang"

index

#获取index
127.0.0.1:6379> lindex n1 0
"wang"

插入

#插入前面
192.168.80.12:6379> linsert names before alex TEST
(integer) 6
192.168.80.12:6379> lrange names 0 -1
1) "zhuangyang"
2) "hanyan"
3) "TEST"
4) "alex"
5) "wangsen"
6) "zouyi"

#插入后面
192.168.80.12:6379> linsert names after alex TEST
(integer) 7
192.168.80.12:6379> lrange names 0 -1
1) "zhuangyang"
2) "hanyan"
3) "TEST"
4) "alex"
5) "TEST"
6) "wangsen"
7) "zouyi"

修改

修改
192.168.80.12:6379> lset names 3 ALEX
OK
192.168.80.12:6379> lrange names 0 -1 
1) "zhuangyang"
2) "hanyan"
3) "TEST"
4) "ALEX"
5) "TEST"
6) "wangsen"
7) "zouyi"

删除多个

192.168.80.12:6379> lrem names 2 TEST
(integer) 2
192.168.80.12:6379> lrange names 0 -1
1) "zhuangyang"
2) "hanyan"
3) "ALEX"
4) "wangsen"
5) "zouyi"

rpoplpush

#从name2 中删除一个值 如果name2为空就等待10秒
192.168.80.12:6379> blpop names2 10
1) "names2"
2) "wangsen"
192.168.80.12:6379> blpop names2 10
(nil) 
(10.02s)

• 集合

#添加集合
192.168.80.12:6379> sadd names3 alex alex jack jack 3 3
(integer) 3
#查看集合
192.168.80.12:6379> smembers names3
1) "alex"
2) "3"
3) "jack"

并集

192.168.80.12:6379> sinter names3 names4  
1) "alex"
2) "jack"

差集

#a 中存在b中不存在的
127.0.0.1:6379> sdiff names3 names4 
1) "jack"

sdiffstore

#把差集加入names2中
127.0.0.1:6379> sdiffstore names2 names3 names4
(integer) 1

查看集合

127.0.0.1:6379> smembers names4
1) "zhang"
2) "san"
3) "han"
4) "wei"
5) "alex"

移动，删除

#把names4的han移到names2
smove names4 names2 han
#删除
srem names4 abc

并集

sunion names4 names3

sismember

#查看一个name是否是集合的成员
192.168.80.12:6379> sismember names3 alex
(integer) 1
192.168.80.12:6379> sismember names3 ale
(integer) 0

• 有序集合

#添加一个有序集合
192.168.80.12:6379> zadd z1 10 alex 5 jack 88 hanyang
(integer) 3

#查看集合
192.168.80.12:6379> zrange z1 0 -1
1) "jack"
2) "alex"
3) "hanyang"

zrank

#查看一个vlaues是否存在
192.168.80.12:6379> zrank z1 alex 
(integer) 1 
192.168.80.12:6379> zrank z1 ajan 
(nil)

zcount

#查看从1-9 
zcount z1 1 9

zrank

获取一个值index
127.0.0.1:6379> zrank z1 han 
(integer) 1

删除

127.0.0.1:6379> zrank z1 han 
(integer) 1

查看并显示编号

127.0.0.1:6379> zrange z9 0 -1 withscores 
1) "han" 
2) "8" 
3) "zhang" 
4) "9" 
5) "alex" 
6) "10" 
7) "zxc" 
8) "88" 
9) "abc" 
10) "100"

• 其他

包括 字典 列表 集合

筛选

127.0.0.1:6379> keys na* 
1) "name" 
2) "names3" 
3) "names2" 
4) "name2" 
5) "names4"

删除

del name4 

#删除设置时间2秒
expire z8 2 

select

切换列表 
select 3

move

#移动位置
127.0.0.1:6379[1]> move name6 2 
(integer) 1 

rename

#改名
127.0.0.1:6379[2]> rename name6 name 
OK 

• 管道

redis-py默认在执行每次请求都会创建（连接池申请连接）和断开（归还连接池）一次连接操作，如果想要在一次请求中指定多个命令，则可以使用pipline实现一次请求指定多个命令，并且默认情况下一次pipline 是原子性操作。

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_
import redis,time

pool = redis.ConnectionPool(host='192.168.1.82',port=6379)

r = redis.Redis(connection_pool=pool)

pipe = r.pipeline()

pipe.set('name','alex')
time.sleep(20)
pipe.set('role','abc')

pipe.execute()


#centos查看
192.168.1.82:6379> get role
"abc"


192.168.1.82:6379> get name
"alex"

• 发布订阅

发布服务器

#!/usr/bin/env python
# _*_ encoding:utf-8 _*_

import redis


class RedisHelper(object):
    def __init__(self):
        self.__conn = redis.Redis(host='192.168.1.82')
        self.chan_sub = 'fm104.5'
        self.chan_pub = 'fm104.5'

    def public(self, msg):
        self.__conn.publish(self.chan_pub,msg)
        return True

    def subscribe(self):
        pub = self.__conn.pubsub()
        pub.subscribe(self.chan_sub)
        pub.parse_response()
        return pub


obj = RedisHelper()
obj.public('hello')

订阅服务器

# _*_coding:utf-8_*_
__author__ = 'Alex Li'

import sys,os

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.append(BASE_DIR)

from day2.monitor import RedisHelper

obj = RedisHelper()
redis_sub = obj.subscribe()

while True:
    msg = redis_sub.parse_response()
    print(msg)