RabbitMq与Redis的使用
<!DOCTYPE html>
css
RabbitMQ 消息队列
python里有threading QUEUE 只用于线程间交互，进程QUEUE 用于父进程与子进程或者是兄弟进程
RabbitMQ采用消息轮询的方式发送消息。一个一个的给每一个消费者
应用之间使用socket实现数据共享
连接几个应用的中间商著名的有：
html
- RabbitMQ
- ZeroMQ
- ActiveMQ
RabbitMQ使用
生产者: - 引用pika模块
import pika - 创建socket
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost')) - 声明一个管道
channel=connection.channel() - 在管道中声明队列
channel.queue_declare(queue='队列名') - 经过管道发送消息 rountingkey就是队列名字
channel.basic_publish(exchange='',routing_key='队列名',body='Hello World!') - 关闭队列,不用关闭管道
connection.close()
消费者(多是其余机器，能够跨机器） - 引用pika模块
import pika - 创建socket
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost')) - 声明一个管道
channel=connection.channel() - 在管道中声明队列
channel.queue_declare(queue='队列名') - 定义函数,标准的处理消息的函数都会带下面四个参数ch管道声明的对象地址，method包含你要发消息给谁的信息，properties包含发消息端的设置信息
def callback(ch,method,properties,body):
print(ch,method,properties,body)
print("[x] Received %r"%body)
ch.basic_ack(delivery_tag=method.delivery_tag)手动确认消息处理完，否则消息一直不销毁 - 消费消息,定义函数的目的，若是收到消息，就用定义的函数处理消息noack参数消息确认，当为True时消息不等消费者确认消息队列就销毁消息，为False时须要等待消费者处理完消息的确认消息队消息队列才销毁消息（判断的是socket是否断开）
channel.basic_consume('定义的函数名',queue='队列名',no_ack=True) - 启动管道接收消息，启动后一直处于开启状态，没有消息就等待。
channel.start_consuming()
RabbitMQ消息分发轮询
采用轮询的方式，依次的发给每一个消费者
生产者会等待消费者肯定处理完消息的回复才会销毁消息。
当消息执行到一半的时候，消费者断开，消息会保留发送给下一个消费者处理
python
消息持久化
消息必须等消费者手动肯定后，才销毁ch.basicack(deliverytag=method.delivery_tag)手动确认消息处理完，否则消息一直不销毁
当RabbitMQ服务中止，服务里的消息队列会销毁。
若是想保持消息队列的持久化，必须在声明队列的时候设置，durable=True。这样当RabbitMQ服务断开再重启，消息队列仍是存在，消息会销毁
channel.queue_declare(queue='队列名'，durable=True)
消息也持久化
channel.basic_publish(exchange='',routing_key='队列名',body='Hello World!'，properties=pika.BasicProperties(delivery_mode=2))
mysql
广播模式
消费者端加channel.basicqos(prefetchcount=1),加过这句话实现了，不是按顺序分发，而是看哪个是空闲的，才分发给空闲的消费者消息。多大本事干多少活。
广播是生产者发消息，全部消费者都收到。
用exchange实现广播。
fanout：全部bind到此exchange的queue均可以接收消息
direct：经过routingkey和exchange决定的那个惟一的queue能够接收消息
topic：全部符合routingkey（此时能够是一个表达式）的routingkey所bind的queue能够接受消息。
git
fanout纯广播
设置管道的时候设置
channel.exchange_declare(exchange='logs',type='fanout')
不声明queue
生产者:
github - 引用pika模块
import pika - 创建socket
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost')) - 声明一个管道
channel=connection.channel() - 设置管道的时候设置
channel.exchange_declare(exchange='logs',exchange_type='fanout') - 经过管道发送消息,广播不须要管道名
channel.basic_publish(exchange='logs',routing_key='',body='Hello World!') - 关闭队列,不用关闭管道
connection.close()
消费者(多是其余机器，能够跨机器） - 引用pika模块
import pika - 创建socket
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost')) - 声明一个管道
channel=connection.channel() - 设置管道的时候设置
channel.exchange_declare(exchange='logs',exchange_type='fanout') - 生成随机queue与exchange转发器绑定。
result=channel.queue<em>declare(exclusive=True)exclusive排他的，单独的，生成随街queue，绑定再exchange上
queue</em>name=result.method.queue<br />
channel.queue<em>bind(exchange='logs',queue=queue</em>name)绑定转发器，接受转发器里的消息，exchange与随机生成的queue绑定
</p>
<p><code>6. 定义函数,标准的处理消息的函数都会带下面四个参数ch管道声明的对象地址，method包含你要发消息给谁的信息，properties</code>
def callback(ch,method,properties,body):
print(ch,method,properties,body)
print("[x] Received %r"%body)<br />
ch.basic<em>ack(delivery</em>tag=method.delivery<em>tag)手动确认消息处理完，否则消息一直不销毁
- 消费消息,定义函数的目的，若是收到消息，就用定义的函数处理消息noack参数消息确认，当为True时消息不等消费者确认消息队列就销毁消息，为False时须要等待消费者处理完消息的确认消息队消息队列才销毁消息（判断的是socket是否断开）
channel.basic_consume('定义的函数名',queue='队列名',no_ack=True) - 启动管道接收消息，启动后一直处于开启状态，没有消息就等待。
channel.start_consuming()
direct广播 info warning error 划分消息
生产者:
web - 引用pika模块
import pika - 创建socket
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost')) - 声明一个管道
channel=connection.channel() - 设置管道的时候设置
channel.exchange_declare(exchange='direct_logs',exchange_type='direct') - 接受分发消息的级别
severity=sys.argv[1] if len(sys.argv)>1 else 'info'
message=' '.join(sys.argv[2:]) or 'hello world!' - 经过管道发送消息,广播不须要管道名
channel.basic_publish(exchange='direct_logs',
routing_key=severity,#相似指定queue
body=message) - 关闭队列,不用关闭管道
connection.close()
消费者 - 引用pika模块
import pika - 创建socket
connection=pika.BlockingConnection(pika.ConnectionParameters('localhost')) - 声明一个管道
channel=connection.channel() - 设置管道的时候设置
channel.exchange_declare(exchange='direct_logs',exchange_type='direct') - 生成随机queue与exchange转发器绑定。
result=channel.queue<em>declare(exclusive=True)exclusive排他的，单独的，生成随街queue，绑定再exchange上
queue</em>name=result.method.queue<br />
severities = sys.argv[1:]
if not severities:
sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])
sys.exit(1)</p>
<p>for severity in severities:
channel.queue<em>bind(exchange='direct</em>logs',queue=queue<em>name,routing</em>key=severity)</p>
<p><code>6. 定义函数,标准的处理消息的函数都会带下面四个参数ch管道声明的对象地址，method包含你要发消息给谁的信息，properties</code>
def callback(ch,method,properties,body):
print(ch,method,properties,body)
print("[x] Received %r"%body)<br />
ch.basic<em>ack(delivery</em>tag=method.delivery<em>tag)手动确认消息处理完，否则消息一直不销毁
- 消费消息,定义函数的目的，若是收到消息，就用定义的函数处理消息noack参数消息确认，当为True时消息不等消费者确认消息队列就销毁消息，为False时须要等待消费者处理完消息的确认消息队消息队列才销毁消息（判断的是socket是否断开）
channel.basic_consume('定义的函数名',queue='队列名',no_ack=True) - 启动管道接收消息，启动后一直处于开启状态，没有消息就等待。
channel.start_consuming()
topic广播
更细致的消息过滤，包括应用程序，#收全部消息，.info接受带有.info的消息，mysql.接受带mysql的消息
生产者
import pika import sys</p> <p>connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel()</p> <p>channel.exchange<em>declare(exchange='topic</em>logs', type='topic')</p> <p>routing<em>key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info' message = ' '.join(sys.argv[2:]) or 'Hello World!' channel.basic</em>publish(exchange='topic<em>logs', routing<em>key=routing</em>key, body=message) print(" [x] Sent %r:%r" % (routing</em>key, message)) connection.close() <code>消费者</code> import pika import sys</p> <p>connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) channel = connection.channel()</p> <p>channel.exchange<em>declare(exchange='topic</em>logs', type='topic')</p> <p>result = channel.queue<em>declare(exclusive=True) queue</em>name = result.method.queue</p> <p>binding<em>keys = sys.argv[1:] if not binding</em>keys: sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0]) sys.exit(1)</p> <p>for binding<em>key in binding</em>keys: channel.queue<em>bind(exchange='topic<em>logs', queue=queue</em>name, routing</em>key=binding_key)</p> <p>print(' [*] Waiting for logs. To exit press CTRL+C') </p> <p>def callback(ch, method, properties, body): print(" [x] %r:%r" % (method.routing_key, body)) </p> <p>channel.basic<em>consume(callback, queue=queue</em>name, no_ack=True) </p> <p>channel.start_consuming() <br />redis
RabbitMQ rpc（remote procedure call）远程调用一个方法
即便生产者又是消费者
startcosuming为阻塞模式，rpc不用阻塞，rpc是执行一会这个再去执行另外一个。processdata_events()非阻塞方法，能收到消息就收，没有消息不阻塞继续往下执行
服务器端
```sql
coding:utf-8
author = 'Alex Li'
import pika
import time
connection = pika.BlockingConnection(pika.ConnectionParameters(
host='localhost'))mongodb
channel = connection.channel()
channel.queuedeclare(queue='rpcqueue')
def fib(n):
if n == 0:
return 0
elif n == 1:
return 1
else:
return fib(n-1) + fib(n-2)
def on_request(ch, method, props, body):
n = int(body)
print(" [.] fib(%s)" % n)
response = fib(n)
ch.basic_publish(exchange='',
routing_key=props.reply_to,
properties=pika.BasicProperties(correlation_id = <br/>
props.correlation_id),
body=str(response))
ch.basic_ack(delivery_tag = method.delivery_tag)
channel.basicqos(prefetchcount=1) channel.basic_consume(onrequest, queue='rpcqueue')
print(" [x] Awaiting RPC requests") channel.start_consuming() 客户端 import pika import uuid
class FibonacciRpcClient(object): def init(self): self.connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost'))
self.channel = self.connection.channel()
result = self.channel.queue_declare(exclusive=True)
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:
self.response = body
def call(self, n):
self.response = None
self.corr_id = str(uuid.uuid4())
self.channel.basic_publish(exchange='',
routing_key='rpc_queue',
properties=pika.BasicProperties(
replyto = self.callbackqueue,#客户端发送消息是，把接收返回消息的管道也告诉给服务器端 correlationid = self.corrid,#用于判断服务器端返回的结果和个人请求是不是同一条。目的就是为了客户端能够同时发两条消息，当服务器端返回结果时，须要有判断关于哪条请求的结果 ), body=str(n)) while self.response is None: self.connection.processdataevents() return int(self.response)
fibonacci_rpc = FibonacciRpcClient()
print(" [x] Requesting fib(30)") response = fibonacci_rpc.call(30) print(" [.] Got %r" % response) ```
Redis
缓存中间商，用socket，例如：mongodb,redis,memcache
* 链接方式
* 链接池
* 操做
- String 操做
- Hash 操做
- List 操做
- Set 操做
- Sort Set 操做
* 管道
* 发布订阅
String操做
- set(name, value, ex=None, px=None, nx=False, xx=False)
在Redis中设置值，默认，不存在则建立，存在则修改
参数：
ex，过时时间（秒）
px，过时时间（毫秒）
nx，若是设置为True，则只有name不存在时，当前set操做才执行
xx，若是设置为True，则只有name存在时，岗前set操做才执行 - setnx(name, value) 设置值，只有name不存在时，执行设置操做（添加）
- setex(name, value, time) 设置值 参数：time，过时时间（数字秒 或 timedelta对象）
- psetex(name, timems, value) 设置值 参数：timems，过时时间（数字毫秒 或 timedelta对象）
- mset(*args, **kwargs) 批量设置值 如：mset(k1='v1', k2='v2')或mget({'k1': 'v1', 'k2': 'v2'})
- get(name) 获取值
- mget(keys, *args) 批量获取 如：mget('ylr', 'wupeiqi')或r.mget(['ylr', 'wupeiqi'])
- getset(name, value) 设置新值并获取原来的值
- getrange(key, start, end) 获取子序列（根据字节获取，非字符） 参数： name，Redis 的 name start，起始位置（字节） end，结束位置（字节） 如： "武沛齐" ，0-3表示 "武"
- setrange(name, offset, value) 修改字符串内容，从指定字符串索引开始向后替换（新值太长时，则向后添加） 参数： offset，字符串的索引，字节（一个汉字三个字节） value，要设置的值
- setbit(name, offset, value) BITCOUNT 统计二进制有多少个1 对name对应值的二进制表示的位进行操做 参数： name，redis的name offset，位的索引（将值变换成二进制后再进行索引） value，值只能是 1 或 0 注：若是在Redis中有一个对应： n1 = "foo"， 那么字符串foo的二进制表示为：01100110 01101111 01101111 因此，若是执行 setbit('n1', 7, 1)，则就会将第7位设置为1， 那么最终二进制则变成 01100111 01101111 01101111，即："goo" 扩展，转换二进制表示： source = "武沛齐" source = "foo" for i in source: num = ord(i) print bin(num).replace('b','') 特别的，若是source是汉字 "武沛齐"怎么办？ 答：对于utf-8，每个汉字占 3 个字节，那么 "武沛齐" 则有 9个字节 对于汉字，for循环时候会按照 字节 迭代，那么在迭代时，将每个字节转换 十进制数，而后再将十进制数转换成二进制 11100110 10101101 10100110 11100110 10110010 10011011 11101001 10111101 10010000 -------------------------- ----------------------------- ----------------------------- 武 沛 齐
- getbit(name, offset) 获取name对应的值的二进制表示中的某位的值 （0或1）
- bitcount(key, start=None, end=None) 获取name对应的值的二进制表示中 1 的个数 参数： key，Redis的name start，位起始位置 end，位结束位置
- bitop(operation, dest, *keys) 获取多个值，并将值作位运算，将最后的结果保存至新的name对应的值 参数： operation,AND（并） 、 OR（或） 、 NOT（非） 、 XOR（异或） dest, 新的Redis的name *keys,要查找的Redis的name 如： bitop("AND", 'newname', 'n1', 'n2', 'n3') 获取Redis中n1,n2,n3对应的值，而后讲全部的值作位运算（求并集），而后将结果保存 newname 对应的值中
- strlen(name) 返回name对应值的字节长度（一个汉字3个字节）
- incr(self, name, amount=1) 自增 name对应的值，当name不存在时，则建立name＝amount，不然，则自增。 参数： name,Redis的name amount,自增数（必须是整数） 注：同incrby
- incrbyfloat(self, name, amount=1.0) 自增 name对应的值，当name不存在时，则建立name＝amount，不然，则自增。 参数： name,Redis的name amount,自增数（浮点型）
- decr(self, name, amount=1) 自减 name对应的值，当name不存在时，则建立name＝amount，不然，则自减。 参数： name,Redis的name amount,自减数（整数）
- append(key, value) 在redis name对应的值后面追加内容 参数： key, redis的name value, 要追加的字符串
- 1. RabbitMq与Redis的使用
- 2. redis与rabbitmq
- 3. RabbitMq与Redis
- 4. Celery与RabbitMQ、Redis
- 5. RabbitMQ的安装与使用
- 6. Rabbitmq的安装与使用
- 7. .NET 环境中使用RabbitMQ RabbitMQ与Redis队列对比 RabbitMQ入门与使用篇
- 8. RabbitMQ与Redis队列对比
- 9. redis集群与rabbitmq集群
- 10. RabbitMQ的使用
- 更多相关文章...
- • 在Spring中使用Redis - Redis教程
- • Redis中使用Lua语言 - Redis教程
- • Composer 安装与使用
- • Java Agent入门实战（一）-Instrumentation介绍与使用
-
每一个你不满意的现在，都有一个你没有努力的曾经。