rabbitmq消息队列

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()

 

 

posted @ 2020-01-02 21:30  凯帅  阅读(...)  评论(...)    收藏