rabbitmq消息队列

Posted 2020-11-24 wukai66

rabbitmq消息队列的学习

1.优点：能够保证消息数据持久化，不丢失

 

安装rabbitmq

1.安装erlang
   $ yum -y install erlang
2.安装RabbitMQ
   $ yum -y install rabbitmq-server
3.启动(无用户名密码):
    systemctl start rabbitmq-server

查看是否启动：

netstat -tunlp

0 0.0.0.0:4369 0.0.0.0:* LISTEN 3134/epmd

4.开启rabbitmq的web控制台

#开启web界面rabbitmq
rabbitmq-plugins enable rabbitmq_management
浏览器访问： http://192.168.91.128:15672/

5.重启rabbitmq

systemctl restart rabbitmq-server

6.创建rabbitmq用户

# 设置新用户kevins 密码123
sudo rabbitmqctl add_user kevins 123

7. 设置用户为admin角色
sudo rabbitmqctl set_user_tags kevins administrator

8.允许kevins用户对所有的队列进行读写

sudo rabbitmqctl set_permissions -p "/" kevins ".*" ".*" ".*"

 

rabbitmq的生产消费者模型

1.准备一个生产者代码，创建队列，向队列中写入消息

python客户端

// rabbitmq官方推荐的python客户端pika模块
pip3 install pika

应用场景1：单发送单接收

生产-消费者模型

P   是生产者
C   是消费者
中间hello是消息队列
可以有多个P、多个C
?
P发送消息给hello队列，C消费者从队列中获取消息，默认轮询方式

生产者send.py

我们的第一个程序send.py将向队列发送一条消息。我们需要做的第一件事是建立与RabbitMQ服务器的连接。

#!/usr/bin/env python
import pika
# 创建凭证，使用rabbitmq用户密码登录
# 去邮局取邮件，必须得验证身份
credentials = pika.PlainCredentials("s14","123")
# 新建连接，这里localhost可以更换为服务器ip
# 找到这个邮局，等于连接上服务器
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.119.10‘,credentials=credentials))
# 创建频道
# 建造一个大邮箱，隶属于这家邮局的邮箱，就是个连接
channel = connection.channel()
# 声明一个队列，用于接收消息，队列名字叫“水许传”
channel.queue_declare(queue=‘水许传‘)
# 注意在rabbitmq中，消息想要发送给队列，必须经过交换(exchange)，初学可以使用空字符串交换(exchange=‘‘)，它允许我们精确的指定发送给哪个队列(routing_key=‘‘),参数body值发送的数据
channel.basic_publish(exchange=‘‘,
                      routing_key=‘水许传‘,
                      body=‘武松又去打老虎啦2‘)
print("已经发送了消息")
# 程序退出前，确保刷新网络缓冲以及消息发送给rabbitmq，需要关闭本次连接
connection.close()

 

可以同时存在多个接受者，等待接收队列的消息，默认是轮训方式分配消息

接受者receive.py，可以运行多次，运行多个消费者

import pika
# 建立与rabbitmq的连接
credentials = pika.PlainCredentials("s14","123")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.119.10‘,credentials=credentials))
channel = connection.channel()
channel.queue_declare(queue="水许传")

def callbak(ch,method,properties,body):
    print("消费者接收到了任务：%r"%body.decode("utf8"))
# 有消息来临，立即执行callbak，没有消息则夯住，等待消息
# 老百姓开始去邮箱取邮件啦，队列名字是水许传
channel.basic_consume("三国演义",callbak,Flase)  # False表示出错后不会丢失
# 开始消费，接收消息
channel.start_consuming()

队列持久化

生产者.py

import pika
# 无密码
# connection = pika.BlockingConnection(pika.ConnectionParameters(‘123.206.16.61‘))
# 有密码
credentials = pika.PlainCredentials("s14","123")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.119.10‘,credentials=credentials))
channel = connection.channel()
# 声明一个队列(创建一个队列)
# 默认此队列不支持持久化，如果服务挂掉，数据丢失
# durable=True 开启持久化，必须新开启一个队列，原本的队列已经不支持持久化了
‘‘‘
实现rabbitmq持久化条件
 delivery_mode=2
使用durable=True声明queue是持久化
 
‘‘‘
channel.queue_declare(queue=‘LOL‘,durable=True)
channel.basic_publish(exchange=‘‘,
                      routing_key=‘LOL‘, # 消息队列名称
                      body=‘德玛西亚万岁‘,
                      # 支持数据持久化
                      properties=pika.BasicProperties(
                          delivery_mode=2,#代表消息是持久的  2
                      )
                      )
connection.close()

消费者：

import pika
credentials = pika.PlainCredentials("s14","123")
connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.119.10‘,credentials=credentials))
channel = connection.channel()
# 确保队列持久化
channel.queue_declare(queue=‘LOL‘,durable=True)

‘‘‘
必须确保给与服务端消息回复，代表我已经消费了数据，否则数据一直持久化，不会消失
‘‘‘
def callback(ch, method, properties, body):
    print("消费者接受到了任务: %r" % body.decode("utf-8"))
    # 模拟代码报错
    # int(‘asdfasdf‘)    # 此处报错，没有给予回复，保证客户端挂掉，数据不丢失
   
    # 告诉服务端，我已经取走了数据，否则数据一直存在
    ch.basic_ack(delivery_tag=method.delivery_tag)
# 关闭no_ack，代表给与回复确认
channel.basic_consume(callback,queue=‘LOL‘,no_ack=False)
channel.start_consuming()