DAY 128 rabbitmq

1 消息队列介绍

1 先进先出”的一种数据机构--》消息队列(mq)
2 MQ解决什么问题
-应用解耦
   -流量消峰
   -消息分发(发布订阅)
   -异步消息(celery:本质是对消息队列的封装)
   
3 主流消息队列产品
-Kafka(互联网公司,数据量大用的多,吞吐量高,数据安全性低一些)
   -rabbitmq(吞吐量低一些,安全性,准确性高)
   

2 rabbitmq安装

1 源码安装
-https://zhuanlan.zhihu.com/p/375157411
2 yum 安装(rpm安装)
# 安装配置epel源
   yum install epel-release -y
   # 安装erlang
   yum -y install erlang
   # 安装RabbitMQ
   yum -y install rabbitmq-server
# 启动
   systemctl start rabbitmq-server
   # 查看是否启动
   ps aux |grep rabbitmq
3 docker安装
-拉一个镜像,启动起来即可
# 安装好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
   # 浏览器访问:
   http://10.0.0.103:15672
   # 输入用户名:admin 密码:admin ,进入到管理控制台

 

3 基本使用(生产者消费者模型)

1 使用python操作rabbitmq
2 pip3 install pika

3.1 生产者

import pika
?
# 无密码
# connection = pika.BlockingConnection(pika.ConnectionParameters(‘101.133.225.166‘))
?
# 有密码
credentials = pika.PlainCredentials("admin","admin")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘101.133.225.166‘,credentials=credentials))
?
?
channel = connection.channel()
?
# 声明一个队列
channel.queue_declare(queue=‘hello‘)
?
channel.basic_publish(exchange=‘‘,
                     routing_key=‘hello‘,
                     body=‘llnb‘)
print("Sent ‘llnb‘")
connection.close()

3.2 消费者

import pika
?
credentials = pika.PlainCredentials("admin","admin")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘101.133.225.166‘,credentials=credentials))
channel = connection.channel()
?
# 声明一个队列(创建一个队列),如果生产者还没放数据,这个队列根本不存,这句话保证代码不出错
channel.queue_declare(queue=‘hello‘)
?
def callback(ch, method, properties, body):
   print("消费者接受到了任务: %r" % body)
   # 通知服务端,消息取走了,如果auto_ack=False,不加下面,消息会一直存在
   # ch.basic_ack(delivery_tag=method.delivery_tag)
?
channel.basic_consume(queue=‘hello‘,on_message_callback=callback,auto_ack=False)
?
channel.start_consuming()

 

4 消息安全之ack

### 生产者
import pika
?
# 无密码
# connection = pika.BlockingConnection(pika.ConnectionParameters(‘101.133.225.166‘))
?
# 有密码
credentials = pika.PlainCredentials("admin","admin")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘101.133.225.166‘,credentials=credentials))
?
?
channel = connection.channel()
?
# 声明一个队列
channel.queue_declare(queue=‘hello‘)
?
channel.basic_publish(exchange=‘‘,
                     routing_key=‘hello‘,
                     body=‘llnb‘)
print("Sent ‘llnb‘")
connection.close()
?
?
### 消费者
import pika
?
credentials = pika.PlainCredentials("admin","admin")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘101.133.225.166‘,credentials=credentials))
channel = connection.channel()
?
# 声明一个队列(创建一个队列),如果生产者还没放数据,这个队列根本不存,这句话保证代码不出错
channel.queue_declare(queue=‘hello‘)
?
def callback(ch, method, properties, body):
   print("消费者接受到了任务: %r" % body)
?
?
   # 通知服务端,消息取走了,如果auto_ack=False,不加下面,消息会一直存在
   # 保证数据安全
   ch.basic_ack(delivery_tag=method.delivery_tag)  # 回复确认,rabbitmq的server就把该消息删除
?
?
# 只要收到消息,立马回复,rabbitmq的server就把消息删除
channel.basic_consume(queue=‘hello‘,on_message_callback=callback,auto_ack=False)
?
channel.start_consuming()

 

DAY 128 rabbitmq

上一篇:npm无法install


下一篇:773. 滑动谜题 力扣(困难) bfs状态改变,写了一个下午