526互联

消息队列Rabbitmq介绍、rabbitmq安装、基于queue实现生产者消费者、基本使用、消息安全(详见笔记)、持久化(详见笔记)、闲置消费(详见笔记)、发布订阅(详见笔记)、发布订阅高级之Routing(按关键字匹配)(详见笔记)

发布时间 2023-05-06 14:36:53作者: 冰柠檬檬

目录

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 闲置消费(详见笔记)

# 正常情况如果有多个消费者，是按照顺序第一个消息给第一个消费者，第二个消息给第二个消费者
# 但是可能第一个消息的消费者处理消息很耗时，一直没结束，就可以让第二个消费者优先获得闲置的消息

8 发布订阅(详见笔记)

9 发布订阅高级之Routing(按关键字匹配)(详见笔记)