RabbitMQ基础总结

2023-08-19 13:00:10

　　MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

　　RabbitMQ和其他AMPQ程序的最大区别在于消息持久化和集群模式，使消息更加可靠。

　　本博客使用的例子均为官方文档中的例子，英文好的可以直接查看原版文档，写此文档的目的作为自己学习过程的一个记录和总结。

本博客测试环境：CentOS6.8VM　　python：2.7.11 操作系统：win7 64位　　rabbitMQ:3.6.5

一、安装

1、yum方式安装

安装配置epel源

   $ rpm -ivh http://dl.fedoraproject.org/pub/epel/6/i386/epel-release-6-8.noarch.rpm

安装erlang

   $ yum -y install erlang

安装RabbitMQ

   $ yum -y install rabbitmq-server

service rabbitmq-server start/stop

2、本地安装

3、安装RabbitMQ的python API

#本次测试使用pika

pip install pika

二、Rabbit简要架构介绍

1、架构图

（图片百度找的，侵删）

2、主要术语介绍

Exchange:交换机，决定了消息路由规则;
Queue:消息队列;
Channel:进行消息读写的通道;
Bind:绑定了Queue和Exchange，意即为符合什么样路由规则的消息，将会放置入哪一个消息队列;

详细的使用方法和具体的概念，相信看完下面的实例会有一个了解。

3、RabbitMQ的主要工作模式

1、简单队列模式，此模Exchange不工作。

2、fanout模式（广播模式），此模式消息在exchange范围内广播。

2、direct模式（精确路由模式），由routing-key精确匹配。

3、topic模式（模糊路由模式），由routing-key模糊匹配

三、简单队列模式

该模式不需要生命Exchange仅使用queue队列来直接交换消息。

生产者

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

# ######################### 生产者 #########################

#建立连接

connection = pika.BlockingConnection(pika.ConnectionParameters(

    host='192.168.79.131'))

#基于连接建立通道

channel = connection.channel()

#创建队列

channel.queue_declare(queue='hello')

#

channel.basic_publish(exchange='',

                      routing_key='hello',

                      body='Hello World!' )

print(" [x] Sent 'Hello World!'")

connection.close()

消费者

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

# ######################### 消费者 #########################

import pika

#创建连接

connection = pika.BlockingConnection(pika.ConnectionParameters(

    host='192.168.79.131'))

#基于连接建立通道

channel = connection.channel()

#创建队列

channel.queue_declare(queue='hello')

#定义消息回调函数

def callback(ch, method, properties, body):

    print(" [x] Received %r" %body)

    time.sleep(5)

    #

channel.basic_consume(callback,

                      queue='hello',

                      no_ack=True)

print(' [c1] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

1、消费者消息应答no-ack

channel.basic_consume(callback,queue='hello',no_ack=True)，默认情况no_ack=False表示consumer完成消息处理后需手动应答。向生队列服务器说明消息已处理完成，可以删除。

如果消费者遇到情况(its channel is closed, connection is closed, or TCP connection is lost)挂掉了，那么，RabbitMQ会重新将该任务添加到队列中，并由其他consumer获取该消息，确保消息不会丢失。

消费者回调函数中ch.basic_ack(delivery_tag=method.delivery_tag)，表示手动应答了服务器。

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

# ######################### 消费者 #########################

import pika

import time

import datetime

#创建连接

connection = pika.BlockingConnection(pika.ConnectionParameters(

    host='192.168.79.131'))

channel = connection.channel()

channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):

    print(" [x] Received %r,%s" %body)

    time.sleep(5)

    ch.basic_ack(delivery_tag=method.delivery_tag)#手动应答

channel.basic_consume(callback,

                      queue='hello',

                      no_ack=False)

print(' [c1] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

4、服务端持久化

RabbitMQ支持消息的持久化，也就是数据写在磁盘上，当我们需要可靠的消息处理的时候应该设置消息持久化。消息队列持久化包括3个部分:

(1)exchange持久化，在声明时指定durable => 1

(2)queue持久化，在声明时指定durable => 1

(3)消息持久化，在投递时指定delivery_mode => 2(1是非持久化)

如果exchange和queue都是持久化的，那么它们之间的binding也是持久化的。如果exchange和queue两者之间有一个持久化，一个非持久化，就不允许建立绑定。

#生产者
#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.79.131'))

channel = connection.channel()

# 持久化队列

channel.queue_declare(queue='hello', durable=True)

channel.basic_publish(exchange='',

                      routing_key='hello',

                      body='Hello World!',

                      properties=pika.BasicProperties(

                          delivery_mode=2, # 消息传递持久化

                      ))

print(" [x] Sent 'Hello World!'")

connection.close()

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.79.131'))

channel = connection.channel()

#队列持久化

channel.queue_declare(queue='hello', durable=True)

def callback(ch, method, properties, body):

    print(" [x] Received %r" % body)

    import time

    time.sleep(10)

    print 'ok'

    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_consume(callback,

                      queue='hello',

                      no_ack=False)

print(' [*] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

消费者

5、消息分发顺序

默认状态下，RabbitMQ将第n个Message分发给第n个Consumer。当然n是取余后的。它不管Consumer是否还有unacked Message，只是按照这个默认机制进行分发。那么如果有个Consumer工作比较重，那么就会导致有的Consumer基本没事可做，有的Consumer却是毫无休息的机会。那么，RabbitMQ是如何处理这种问题呢？

通过 basic.qos 方法设置prefetch_count=1 。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理一个Message。换句话说，在接收到该Consumer的ack前，他它不会将新的Message分发给它。

6、最终版本

设置消息确认，服务器的持久化，和消息分发顺序后，最终版本：

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.79.131'))

channel = connection.channel()

# make message persistent

channel.queue_declare(queue='hello')

def callback(ch, method, properties, body):

    print(" [x] Received %r" % body)

    import time

    time.sleep(10)

    print 'ok'

    ch.basic_ack(delivery_tag = method.delivery_tag)

channel.basic_qos(prefetch_count=1)

channel.basic_consume(callback,

                      queue='hello',

                      no_ack=False)

print(' [*] Waiting for messages. To exit press CTRL+C')

channel.start_consuming()

消费者

#生产者

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(host='192.168.79.131'))

channel = connection.channel()

# 持久化队列

channel.queue_declare(queue='hello', durable=True)

channel.basic_publish(exchange='',

                      routing_key='hello',

                      body='Hello World!',

                      properties=pika.BasicProperties(

                          delivery_mode=2, # 消息传递持久化

                      ))

print(" [x] Sent 'Hello World!'")

connection.close()

生产者

四、fanout-广播模式

广播模式也称为发布订阅和简单的消息队列区别在于，发布订阅会将消息发送给所有的订阅者，而消息队列中的数据被消费一次便消失。所以，RabbitMQ实现发布和订阅时，会为每一个订阅者创建一个队列，而发布者发布消息时，会将消息放置在所有相关队列中。

关键点：

　　1、生产者：设置exchange type = fanout

　　2、消费者：result = channel.queue_declare(exclusive=True)　　#生成临时队列，当Consumer关闭连接时，这个queue要被deleted。

　　3、消费者：queue_name = result.method.queue　　#result.method.queue 可以取得queue的名字。基本上都是这个样子：amq.gen-JzTY20BRgKO-HjmUJxsgf。

下面我们模拟一个简单的日志系统，将所有日志简单的广播给所有订阅者。

生产者：

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='localhost'))

channel = connection.channel()

channel.exchange_declare(exchange='logs',

                         type='fanout')

message = ' '.join(sys.argv[1:]) or "info: Hello World!"

channel.basic_publish(exchange='logs',

                      routing_key='',

                      body=message)

print(" [x] Sent %r" % message)

connection.close()

消费者：

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='localhost'))

channel = connection.channel()

channel.exchange_declare(exchange='logs',

                         type='fanout')

result = channel.queue_declare(exclusive=True)　　#

queue_name = result.method.queue

channel.queue_bind(exchange='logs',

                   queue=queue_name)

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):

    print(" [x] %r" % body)

channel.basic_consume(callback,

                      queue=queue_name,

                      no_ack=True)

channel.start_consuming()

运行消费者两个消费者和一个生产者，多个消费者均收到消息

消费者开启：queue和binding状态

[root@localhost ~]# rabbitmqctl list_bindings

Listing bindings ...

    　　exchange    amq.gen-5A-Amc-4DZgaE5_x52Q-Tw    queue    amq.gen-5A-Amc-4DZgaE5_x52Q-Tw    []

    　　exchange    amq.gen-wGFWTvv42rUR6_tki5eeOg    queue    amq.gen-wGFWTvv42rUR6_tki5eeOg    []

logs    exchange    amq.gen-5A-Amc-4DZgaE5_x52Q-Tw    queue    amq.gen-5A-Amc-4DZgaE5_x52Q-Tw    []

logs    exchange    amq.gen-wGFWTvv42rUR6_tki5eeOg    queue    amq.gen-wGFWTvv42rUR6_tki5eeOg    []

[root@localhost ~]# rabbitmqctl list_queues

Listing queues ...

amq.gen-5A-Amc-4DZgaE5_x52Q-Tw    0

amq.gen-wGFWTvv42rUR6_tki5eeOg    0

关闭消费者：queue和bindings均为空。

[root@localhost ~]# rabbitmqctl list_bindings

Listing bindings ...

...done.

[root@localhost ~]# rabbitmqctl list_queues

Listing queues ...

...done.

五、direct模式-精确路由模式

Direct exchange的路由，通过routing_key（我将其称为路由关键字）的精确匹配。即时一个queue的两条routing_key同时匹配到同一条消息，也仅接收一条。

上个主题我们实现的简单的日志广播系统。这个小节我们实现更复杂点的需求，针对不同的日志等级发送给不同的订阅者。

基本原理就是：

　　1、生产者：在fanout模式的基础上，配置routing_key。

　　2、消费者：在fanout模式的基础上，配置routing_key。

下面我们模拟一个简单的日志系统，将所有日志简单的广播给所有订阅者。

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

#!/usr/bin/env python

import pika

import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='192.168.79.131'))

channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',

                         type='direct')

result = channel.queue_declare(exclusive=True)

queue_name = result.method.queue

severities = sys.argv[1:]

if not severities:

    sys.stderr.write("Usage: %s [info] [warning] [error]\n" % sys.argv[0])

    sys.exit(1)

#使用for循环来同时绑定多个binding-key

for severity in severities:

    channel.queue_bind(exchange='direct_logs',

                       queue=queue_name,

                       routing_key=severity)

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):

    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(callback,

                      queue=queue_name,

                      no_ack=True)

channel.start_consuming()

消费者

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='192.168.79.131'))

channel = connection.channel()

channel.exchange_declare(exchange='direct_logs',

                         type='direct')

severity = sys.argv[1] if len(sys.argv) > 1 else 'info'

message = ' '.join(sys.argv[2:]) or 'Hello World!'

channel.basic_publish(exchange='direct_logs',

                      routing_key=severity,

                      body=message)

print(" [x] Sent %r:%r" % (severity, message))

connection.close()

生产者

关于exchange和queue的一对多或者多对一匹配：

　　1、消费者的一个queue可以绑定多个bindding_key来同时匹配多个路由关键字。使用循环把多个routing_key绑定到一个queue上。

　　2、多个queue可以匹配同一个routing_key来匹配同一个路由关键字。当所有queue的routing_key相同时，模式就和fanout一样啦。

六、topic模式-模糊路由匹配

topic模式基本与Direct模式相同，唯一的不同点就是routing-key是模糊匹配。routing_key的格式以点号分隔，最长255bytes。

匹配元字符：

　　# 表示可以匹配 0 个或任意个单词(包含1个)

　　* 表示只能匹配一个单词

Topic exchange和其他exchange：

    如果binding_key 是#   ，那么它会接收所有的Message，不管routing_key是什么，就像是fanout exchange。

    如果 #和* 没有被使用，那么topic exchange就变成了direct exchange。

发送者路由值           roiting_key

i.love.python          i.*  -- 不匹配

i.love.python          i.#  -- 匹配

i.love.python　　　　　　#　　--匹配，单个#匹配所有

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='192.168.79.131'))

channel = connection.channel()

channel.exchange_declare(exchange='topic_logs',

                         type='topic')

result = channel.queue_declare(exclusive=True)

queue_name = result.method.queue

binding_keys = sys.argv[1:]

if not binding_keys:

    sys.stderr.write("Usage: %s [binding_key]...\n" % sys.argv[0])

    sys.exit(1)

for binding_key in binding_keys:

    channel.queue_bind(exchange='topic_logs',

                       queue=queue_name,

                       routing_key=binding_key)

print(' [*] Waiting for logs. To exit press CTRL+C')

def callback(ch, method, properties, body):

    print(" [x] %r:%r" % (method.routing_key, body))

channel.basic_consume(callback,

                      queue=queue_name,

                      no_ack=True)

channel.start_consuming()

消费者

#! /usr/bin/env python

# -*- coding:utf-8 -*-

# __author__ = "TKQ"

import pika

import sys

connection = pika.BlockingConnection(pika.ConnectionParameters(

        host='192.168.79.131'))

channel = connection.channel()

channel.exchange_declare(exchange='topic_logs',

                         type='topic')

routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'

message = ' '.join(sys.argv[2:]) or 'Hello World!'

channel.basic_publish(exchange='topic_logs',

                      routing_key=routing_key,

                      body=message)

print(" [x] Sent %r:%r" % (routing_key, message))

connection.close()

生产者

码农公寓