rabbitMq

参考项目

使用docker安装

1. 查看rabbitmq资源

docker search rabbitmq:management

2. 拉取rabbitmq

docker pull rabbitmq:management

3. 运行

docker run -di --name=tensquare_rabbitmq -p 5672:5672 -p 5671:5671 -p 4369:4369 -p 15671:15671 -p 15672:15672 -p 25672:25672 rabbitmq:managementd

4. 端口介绍

15671 management监听端口
15672 web 登录
5672,5671 （AMQP 0-9-1 without and with TLS）
4369 epmd 代表 Erlang 端口映射守护进程
25672 （Erlang distribution）

5.访问

42.193.152.252:15672 访问web
账号:guest
密码:guest

6.项目中使用，添加pom

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
<dependency>

添加用户

创建账号
    rabbitmqctl add_user admin 123

设置用户管理员
    rabbitmqctl set_user_tags admin administrator

设置用户权限
    set_permissions [-p <vhostpath>] <user> <conf> <write> <read>
    rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

python使用

配置信息

1.queue队列

2.exchang交换机

持久化配置

RabbitMq 持久化

MQ默认建立的是临时 queue 和 exchange，如果不声明持久化，一旦 rabbitmq 挂掉，queue、exchange 将会全部丢失。所以我们一般在创建 queue 或者 exchange 的时候会声明持久化。 1.queue 声明持久化

# 声明消息队列，消息将在这个队列传递，如不存在，则创建。durable = True 代表消息队列持久化存储，False 非持久化存储 
result = channel.queue_declare(queue = 'python-test',durable = True)

2.exchange 声明持久化

# 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建.durable = True 代表exchange持久化存储，False 非持久化存储 
channel.exchange_declare(exchange = 'python-test', durable = True)

注意：如果已存在一个非持久化的 queue 或 exchange ，执行上述代码会报错，因为当前状态不能更改 queue 或 exchange 存储属性，需要删除重建。如果 queue 和 exchange 中一个声明了持久化，另一个没有声明持久化，则不允许绑定。 3.消息持久化虽然 exchange 和 queue 都申明了持久化，但如果消息只存在内存里，rabbitmq 重启后，内存里的东西还是会丢失。所以必须声明消息也是持久化，从内存转存到硬盘。

# 向队列插入数值 routing_key是队列名。delivery_mode = 2 声明消息在队列中持久化，delivery_mod = 1 消息非持久化
channel.basic_publish(exchange = '',routing_key = 'python-test',body = message,
                      properties=pika.BasicProperties(delivery_mode = 2))

4.acknowledgement 消息不丢失消费者（consumer）调用callback函数时，会存在处理消息失败的风险，如果处理失败，则消息丢失。但是也可以选择消费者处理失败时，将消息回退给 rabbitmq ，重新再被消费者消费，这个时候需要设置确认标识。

# no_ack 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列。True，无论调用callback成功与否，消息都被消费掉
channel.basic_consume(callback,queue = 'python-test',auto_ack = False)

1.普通消息发送和接受

根据消费者连接mq的顺序，轮询

假如client_1, client_2, client_3三个消费者连接此队列 server生产者发布了5个任务

client_1: 收到任务 1，4
client_2: 收到任务 2，5
client_3: 收到任务 3

队列消息的产生者，负责生产消息，并将消息传入队列消费者（consumer）：队列消息的接收者，负责接收并处理消息队列中的消息

生产者发送消息, 等待消费, 否则一直存在

import pika
import json

credentials = pika.PlainCredentials(username='admin', password='CHENHAIFENG')  # mq用户名和密码
# 虚拟队列需要指定参数 virtual_host，如果是默认的可以不填。
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.50.225', port=5672, virtual_host='diary', credentials=credentials))
channel = connection.channel()  # 声明管道

# 声明消息队列，消息将在这个队列中进行传递。如果将消息发送到不存在的队列，rabbitmq将会自动清除这些消息。如果队列不存在，则创建
# durable=True 虽然有了消息反馈机制，但是如果rabbitmq自身挂掉的话，那么任务还是会丢失。所以需要将任务持久化存储起来。声明持久化存储
result = channel.queue_declare(queue='python-test', durable=True)

for i in range(5):
    message = json.dumps({'OrderId': "1000%s" % i})
    # routing_key是队列名
    # exchange表示交换器，能精确指定消息应该发送到哪个队列(有指定的队列可以不进行配置)，
    # routing_key设置为队列的名称
    # body就是发送的内容
    # properties 持久化
    channel.basic_publish(exchange='', routing_key='python-test', body=message,
                          properties=pika.BasicProperties(delivery_mode=2))
    print(message)

connection.close()

消费者消费

import pika

credentials = pika.PlainCredentials('admin', 'CHENHAIFENG')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.50.225', port=5672, virtual_host='diary', credentials=credentials))
channel = connection.channel()
# 申明消息队列，消息在这个队列传递，如果不存在，则创建队列
channel.queue_declare(queue='python-test', durable=True)


# 定义一个回调函数来处理消息队列中的消息，这里是打印出来
def callback(ch, method, properties, body):
    # exit()  # 在这里exit，就会中断发送确认标识，可以用来测试消息确认功能
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(body.decode())
    print(method.delivery_tag) # 为消费次数


# queue 队列
# callback 接收消息
# no_ack=False（默认为False，即必须有确认标识）,在回调函数consumer_callback中，未收到确认标识，那么，RabbitMQ会重新将该任务添加到队列中。
channel.basic_consume(queue='python-test', on_message_callback=callback, auto_ack=False)

# 开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理
channel.start_consuming()

2.模式fanout

这种模式下，传递到 exchange 的消息将会转发到所有与其绑定的 queue 上。

不需要指定 routing_key ，即使指定了也是无效。
需要提前将 exchange 和 queue 绑定，一个 exchange 可以绑定多个 queue，一个queue可以绑定多个exchange。
需要先启动订阅者，此模式下的队列是 consumer 随机生成的，发布者仅仅发布消息到 exchange ，由 exchange 转发消息至 queue。
生产者生成后, 有订阅者在则消费,没有则马上消失

# rabbitmq 发布者
import pika

# 创建连接
credentials = pika.PlainCredentials('admin', 'CHENHAIFENG')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.50.225', port=5672, virtual_host='diary', credentials=credentials))
channel = connection.channel()
# 定义交换机，exchange表示交换机名称，type表示类型
channel.exchange_declare(exchange='diary_fanout',
                         exchange_type='fanout',
                         durable=True)


# 将消息发送到交换机
for i in range(10):
    message = bytes('Hello Python-%s' % i, encoding='utf8')
    channel.basic_publish(exchange='diary_fanout',  # 指定exchange
                          routing_key='diary',  # fanout下不需要配置，配置了也不会生效
                          body=message,
                          properties=pika.BasicProperties(delivery_mode=2))

connection.close()

消费者等待

#  rabbitmq 订阅者
import pika

credentials = pika.PlainCredentials('admin', 'CHENHAIFENG')
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='192.168.50.225', port=5672, virtual_host='diary', credentials=credentials))
channel = connection.channel()

# 定义交换机，进行exchange声明，exchange表示交换机名称，type表示类型
channel.exchange_declare(exchange='diary_fanout',
                         exchange_type='fanout',
                         durable=True)

# 随机创建队列
# result = channel.queue_declare(queue='diary', exclusive=True)  # exclusive=True表示建立临时队列，当consumer关闭后，该队列就会被删除
# queue_name = result.method.queue

# 将队列与exchange进行绑定
channel.queue_bind(exchange='diary_fanout',
                   queue='python-test')


def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag=method.delivery_tag)
    print(" [x] %r" % body)
    # print(method.delivery_tag)


# 从队列获取信息
channel.basic_consume(on_message_callback=callback,
                      queue='python-test',
                      auto_ack=False)

channel.start_consuming()

3.模式direct

这种工作模式的原理是消息发送至 exchange，exchange 根据 路由键（routing_key）转发到相对应的 queue 上。

可以使用默认 exchange =' ' ，也可以自定义 exchange
这种模式下不需要将 exchange 和任何进行绑定，当然绑定也是可以的。可以将 exchange 和 queue ，routing_key 和 queue 进行绑定
传递或接受消息时需要 指定 routing_key
需要先启动 订阅者，此模式下的队列是 consumer 随机生成的，发布者 仅仅发布消息到 exchange ，由 exchange 转发消息至 queue。

import pika
import json

credentials = pika.PlainCredentials('shampoo', '123456')  # mq用户名和密码
# 虚拟队列需要指定参数 virtual_host，如果是默认的可以不填。
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel=connection.channel()
# 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='direct')

for i in range(10):
    message=json.dumps({'OrderId':"1000%s"%i})
# 指定 routing_key。delivery_mode = 2 声明消息在队列中持久化，delivery_mod = 1 消息非持久化
    channel.basic_publish(exchange = 'python-test',routing_key = 'OrderId',body = message,
                          properties=pika.BasicProperties(delivery_mode = 2))
    print(message)
connection.close()

import pika

credentials = pika.PlainCredentials('shampoo', '123456')
connection = pika.BlockingConnection(pika.ConnectionParameters(host = '10.1.62.170',port = 5672,virtual_host = '/',credentials = credentials))
channel = connection.channel()
# 创建临时队列，队列名传空字符，consumer关闭后，队列自动删除
result = channel.queue_declare('',exclusive=True)
# 声明exchange，由exchange指定消息在哪个队列传递，如不存在，则创建。durable = True 代表exchange持久化存储，False 非持久化存储
channel.exchange_declare(exchange = 'python-test',durable = True, exchange_type='direct')
# 绑定exchange和队列  exchange 使我们能够确切地指定消息应该到哪个队列去
channel.queue_bind(exchange = 'python-test',queue = result.method.queue,routing_key='OrderId')
# 定义一个回调函数来处理消息队列中的消息，这里是打印出来
def callback(ch, method, properties, body):
    ch.basic_ack(delivery_tag = method.delivery_tag)
    print(body.decode())


#channel.basic_qos(prefetch_count=1)
# 告诉rabbitmq，用callback来接受消息
channel.basic_consume(result.method.queue,callback,
# 设置成 False，在调用callback函数时，未收到确认标识，消息会重回队列。True，无论调用callback成功与否，消息都被消费掉
                      auto_ack = False)
channel.start_consuming()

4.topicd

这种模式和第二种模式差不多，exchange 也是通过路由键 routing_key 来转发消息到指定的 queue 。不同点是 routing_key 使用正则表达式支持模糊匹配，但匹配规则又与常规的正则表达式不同，比如“#”是匹配全部，“*”是匹配一个词。举例：routing_key =“#orderid#”，意思是将消息转发至所有 routing_key 包含 “orderid” 字符的队列中。代码和模式二类似，就不贴出来了。

总结

fanout与direct ---direct生产者key发布, 订阅者根据key消费

CHF Blog ~

ChenBlog

[ rabbitmq基础 ]