目录
- 1 消息队列Rabbitmq介绍
- 2 rabbitmq安装
- 3 基于queue实现生产者消费者
- 4 基本使用
- 5 消息安全(详见笔记)
- 6 持久化(详见笔记)
- 7 闲置消费(详见笔记)
- 8 发布订阅(详见笔记)
- 9 发布订阅高级之Routing(按关键字匹配)(详见笔记)
1 消息队列Rabbitmq介绍
# 消息队列
-中间件---（概念很大）---》准确一些叫消息队列中间件
-消息队列中间件
-使用redis当做消息队列来用
-blpop 阻塞式弹出，实现队列，先进先出
# MQ ,消息队列，MessageQueue 是什么
消息队列就是基础数据结构中的“先进先出”的一种数据机构。想一下，生活中买东西，需要排队，先排的人先买消费，就是典型的“先进先出”
# MQ解决什么问题
-应用解耦
-流量削峰
-消息分发（发布订阅）
-异步消息
-IPC 进程间通信也可以通过消息队列
#
2 rabbitmq安装
# win:https://www.rabbitmq.com/install-windows-manual.html
-erlang解释器
-rabbitmq的软件
# centos
yum -y install erlang
yum -y install rabbitmq-server
# docker安装
docker pull rabbitmq:management
docker run -di --name Myrabbitmq -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin -p 15672:15672 -p 5672:5672 rabbitmq:management
# 访问虚拟机的15672端口，就可以看到图形化界面（官方提供的），手动点点点操作
http://10.0.0.101:15672/#/
3 基于queue实现生产者消费者
import Queue
import threading
message = Queue.Queue(10)
def producer(i):
while True:
message.put(i)
def consumer(i):
while True:
msg = message.get()
for i in range(12):
t = threading.Thread(target=producer, args=(i,))
t.start()
for i in range(10):
t = threading.Thread(target=consumer, args=(i,))
t.start()
4 基本使用
4.1 发送者
import pika
###第一步,连接
# 无密码
# connection = pika.BlockingConnection(pika.ConnectionParameters(host='10.0.0.101',port=5672))
# 有密码
credentials = pika.PlainCredentials("admin","admin")
connection = pika.BlockingConnection(pika.ConnectionParameters('10.0.0.101',credentials=credentials))
# 第二步:连接channal
channel = connection.channel()
# 第三步:创建一个队列,名字叫hello
channel.queue_declare(queue='hello')
# 第三步:向hello队列中,发送Hello World
# routing_key 队列名字
#body 发送的Neri
channel.basic_publish(exchange='', routing_key='hello', body='Hello World4441!')
print(" Sent 'Hello World!'")
connection.close()
4.2 消费者
import pika
# 连接rabbitmq
credentials = pika.PlainCredentials("admin", "admin")
connection = pika.BlockingConnection(pika.ConnectionParameters('10.0.0.101', credentials=credentials))
#假设队列不存在,创建
channel = connection.channel()
channel.queue_declare(queue='hello')
# 回调函数
def callback(ch, method, properties, body):
print(" [x] Received %r" % body)
channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming() # 程序会夯在这里，等待从消息队列中取消息
5 消息安全(详见笔记)
# 消费完，确认后，再删除消息
6 持久化(详见笔记)
# queue和消息都要持久化
7 闲置消费(详见笔记)
# 正常情况如果有多个消费者，是按照顺序第一个消息给第一个消费者，第二个消息给第二个消费者
# 但是可能第一个消息的消费者处理消息很耗时，一直没结束，就可以让第二个消费者优先获得闲置的消息