博客

python rabbitmq callback 传参数_python – 操作RabbitMQ

  • Post author:
  • Post category:python


介绍

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统。他遵循Mozilla Public License开源协议。

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

应用场景：

RabbitMQ无疑是目前最流行的消息队列之一，对各种语言环境的支持也很丰富，作为一个.NET developer有必要学习和了解这一工具。消息队列的使用场景大概有3种：

1、系统集成，分布式系统的设计。各种子系统通过消息来对接，这种解决方案也逐步发展成一种架构风格，即“通过消息传递的架构”。

2、当系统中的同步处理方式严重影响了吞吐量，比如日志记录。假如需要记录系统中所有的用户行为日志，如果通过同步的方式记录日志势必会影响系统的响应速度，当我们将日志消息发送到消息队列，记录日志的子系统就会通过异步的方式去消费日志消息。

3、系统的高可用性，比如电商的秒杀场景。当某一时刻应用服务器或数据库服务器收到大量请求，将会出现系统宕机。如果能够将请求转发到消息队列，再由服务器去消费这些消息将会使得请求变得平稳，提高系统的可用性。

安装

安装RabbitMQ

48304ba5e6f9fe08f3fa1abda7d326ab.png

基础环境：

内核

3.10.0-327.el7.x86_64

系统版本

CentOS Linux release 7.2.1511 (Core)

安装配置epel源

# rpm -ivh http://mirrors.neusoft.edu.cn/epel/7/x86_64/e/epel-release-7-7.noarch.rpm

安装erlang

# yum install erlang

下载RabbitMQ 3.6.1

# wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-3.6.1-1.noarch.rpm

安装rabbitmq-server

# rpm -ivh rabbitmq-server-3.6.1-1.noarch.rpm

生成配置文件

# cp /usr/share/doc/rabbitmq-server-3.6.1/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

启动RabbitMQ

# rabbitmq-server start

48304ba5e6f9fe08f3fa1abda7d326ab.png

安装Python API

# pip3 install pika

or

# easy_install pika

Python 操作RabbitMQ

对于RabbitMQ来说，生产和消费不再针对内存里的一个Queue对象，而是某台服务器上的RabbitMQ Server实现的消息队列。

1.生产者代码

48304ba5e6f9fe08f3fa1abda7d326ab.png

#!/usr/bin/env python

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

# auth : pangguoping

import pika

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

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

#链接rabbit服务器(localhost是本机，如果是其他服务器请修改为ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

#创建频道

channel = connection.channel()

# 声明消息队列，消息将在这个队列中进行传递。如果将消息发送到不存在的队列，rabbitmq将会自动清除这些消息。如果队列不存在，则创建

channel.queue_declare(queue=’hello’)

#exchange — 它使我们能够确切地指定消息应该到哪个队列去。

#向队列插入数值 routing_key是队列名 body是要插入的内容

channel.basic_publish(exchange=”,

routing_key=’hello’,

body=’Hello World!’)

print(“开始队列”)

#缓冲区已经flush而且消息已经确认发送到了RabbitMQ中，关闭链接

connection.close()

48304ba5e6f9fe08f3fa1abda7d326ab.png

2.消费者代码

48304ba5e6f9fe08f3fa1abda7d326ab.png

#!/usr/bin/env python

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

# auth : pangguoping

import pika

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

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

# 连接到rabbitmq服务器

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

channel = connection.channel()

# 声明消息队列，消息将在这个队列中进行传递。如果队列不存在，则创建

channel.queue_declare(queue=’wzg’)

# 定义一个回调函数来处理，这边的回调函数就是将信息打印出来。

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

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

# 告诉rabbitmq使用callback来接收信息

channel.basic_consume(callback,

queue=’hello’,

no_ack=True)

# no_ack=True表示在回调函数中不需要发送确认标识

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

# 开始接收信息，并进入阻塞状态，队列里有信息才会调用callback进行处理。按ctrl+c退出。

channel.start_consuming()

48304ba5e6f9fe08f3fa1abda7d326ab.png

RabbitMQ持久化

1、acknowledgment 消息不丢失的方法

生效方法：channel.basic_consume(consumer_callback, queue, no_ack=False, exclusive=False, consumer_tag=None, arguments=None)

即no_ack=False(默认为False，即必须有确认标识),在回调函数consumer_callback中，未收到确认标识，那么，RabbitMQ会重新将该任务添加到队列中。

生产者代码

48304ba5e6f9fe08f3fa1abda7d326ab.png

#!/usr/bin/env python

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

# auth : pangguoping

import pika

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

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

#链接rabbit服务器(localhost是本机，如果是其他服务器请修改为ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

#创建频道

channel = connection.channel()

# 声明消息队列，消息将在这个队列中进行传递。如果将消息发送到不存在的队列，rabbitmq将会自动清除这些消息。如果队列不存在，则创建

channel.queue_declare(queue=’hello’)

#exchange — 它使我们能够确切地指定消息应该到哪个队列去。

#向队列插入数值 routing_key是队列名 body是要插入的内容

channel.basic_publish(exchange=”,

routing_key=’hello’,

body=’Hello World!’)

print(“开始队列”)

#缓冲区已经flush而且消息已经确认发送到了RabbitMQ中，关闭链接

connection.close()

48304ba5e6f9fe08f3fa1abda7d326ab.png

消费者代码：

48304ba5e6f9fe08f3fa1abda7d326ab.png

import pika

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

# 链接rabbit

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

# 创建频道

channel = connection.channel()

# 如果生产者没有运行创建队列，那么消费者创建队列

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_consume(callback,

queue=’hello’,

no_ack=False)

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

channel.start_consuming()

48304ba5e6f9fe08f3fa1abda7d326ab.png

2、消息持久化存储(Message durability)

虽然有了消息反馈机制，但是如果rabbitmq自身挂掉的话，那么任务还是会丢失。所以需要将任务持久化存储起来。声明持久化存储

channel.queue_declare(queue=’wzg’, durable=True) # 声明队列持久化

Ps: 但是这样程序会执行错误，因为‘wzg’这个队列已经存在，并且是非持久化的，rabbitmq不允许使用不同的参数来重新定义存在的队列。因此需要重新定义一个队列

channel.queue_declare(queue=’test_queue’, durable=True) # 声明队列持久化

注意：如果仅仅是设置了队列的持久化，仅队列本身可以在rabbit-server宕机后保留，队列中的信息依然会丢失，如果想让队列中的信息或者任务保留，还需要做以下设置：

48304ba5e6f9fe08f3fa1abda7d326ab.png

channel.basic_publish(exchange=”,

routing_key=”test_queue”,

body=message,

properties=pika.BasicProperties(

delivery_mode = 2, # 使消息或任务也持久化存储

))

48304ba5e6f9fe08f3fa1abda7d326ab.png

48304ba5e6f9fe08f3fa1abda7d326ab.png

消息队列持久化包括3个部分：

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

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

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

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

48304ba5e6f9fe08f3fa1abda7d326ab.png

发布与订阅

RabbitMQ的发布与订阅，借助于交换机(Exchange)来实现。

交换机的工作原理：消息发送端先将消息发送给交换机，交换机再将消息发送到绑定的消息队列，而后每个接收端(consumer)都能从各自的消息队列里接收到信息。

6a70f268cc117077f2aa743905d704c5.png

Exchange有三种工作模式，分别为：Fanout, Direct, Topic

模式1 Fanout

任何发送到Fanout Exchange的消息都会被转发到与该Exchange绑定(Binding)的所有Queue上

1.可以理解为路由表的模式

2.这种模式不需要routing_key(即使指定，也是无效的)

3.这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。

4.如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。

注意：这个时候必须先启动消费者，即订阅者。因为随机队列是在consumer启动的时候随机生成的，并且进行绑定的。producer仅仅是发送至exchange，并不直接与随机队列进行通信。

生产者代码

48304ba5e6f9fe08f3fa1abda7d326ab.png

#!/usr/bin/env python

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

# auth : pangguoping

# rabbitmq 发布者

import pika

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

#链接rabbit服务器(localhost是本机，如果是其他服务器请修改为ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

channel = connection.channel()

# 定义交换机，exchange表示交换机名称，type表示类型

channel.exchange_declare(exchange=’logs_fanout’,

type=’fanout’)

message = ‘Hello Python’

# 将消息发送到交换机

channel.basic_publish(exchange=’logs_fanout’, # 指定exchange

routing_key=”, # fanout下不需要配置，配置了也不会生效

body=message)

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

connection.close()

48304ba5e6f9fe08f3fa1abda7d326ab.png

消费者代码

48304ba5e6f9fe08f3fa1abda7d326ab.png

#!/usr/bin/env python

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

# auth : pangguoping

# rabbitmq 订阅者

import pika

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

#链接rabbit服务器(localhost是本机，如果是其他服务器请修改为ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

channel = connection.channel()

# 定义交换机，进行exchange声明，exchange表示交换机名称，type表示类型

channel.exchange_declare(exchange=’logs_fanout’,

type=’fanout’)

# 随机创建队列

result = channel.queue_declare(exclusive=True) # exclusive=True表示建立临时队列，当consumer关闭后，该队列就会被删除

queue_name = result.method.queue

# 将队列与exchange进行绑定

channel.queue_bind(exchange=’logs_fanout’,

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

48304ba5e6f9fe08f3fa1abda7d326ab.png

模式2  Direct

e233da8ca63a5c65abae43ae8ef49747.png

路由键的工作原理：每个接收端的消息队列在绑定交换机的时候，可以设定相应的路由键。发送端通过交换机发送信息时，可以指明路由键 ，交换机会根据路由键把消息发送到相应的消息队列，这样接收端就能接收到消息了。

任何发送到Direct Exchange的消息都会被转发到routing_key中指定的Queue：

1.一般情况可以使用rabbitMQ自带的Exchange：””  (该Exchange的名字为空字符串)， 也可以自定义Exchange

2.这种模式下不需要将Exchange进行任何绑定(bind)操作。当然也可以进行绑定。可以将不同的routing_key与不同的queue进行绑定，不同的queue与不同exchange进行绑定

3.消息传递时需要一个“routing_key”

4.如果消息中中不存在routing_key中绑定的队列名，则该消息会被抛弃。

如果一个exchange 声明为direct，并且bind中指定了routing_key,那么发送消息时需要同时指明该exchange和routing_key.

消费者代码

48304ba5e6f9fe08f3fa1abda7d326ab.png

#!/usr/bin/env python

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

# auth : pangguoping

# 消费者

import pika

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

#链接rabbit服务器(localhost是本机，如果是其他服务器请修改为ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

channel = connection.channel()

# 定义exchange和类型

channel.exchange_declare(exchange=’direct_test’,

type=’direct’)

# 生成随机队列

result = channel.queue_declare(exclusive=True)

queue_name = result.method.queue

severities = [‘error’, ]

# 将随机队列与routing_key关键字以及exchange进行绑定

for severity in severities:

channel.queue_bind(exchange=’direct_test’,

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

48304ba5e6f9fe08f3fa1abda7d326ab.png

生产者

48304ba5e6f9fe08f3fa1abda7d326ab.png

#!/usr/bin/env python

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

# auth : pangguoping

# 发布者

import pika

credentials = pika.PlainCredentials(‘admin’, ‘admin’)

#链接rabbit服务器(localhost是本机，如果是其他服务器请修改为ip地址)

connection = pika.BlockingConnection(pika.ConnectionParameters(‘192.168.1.103′,5672,’/’,credentials))

channel = connection.channel()

# 定义交换机名称及类型

channel.exchange_declare(exchange=’direct_test’,

type=’direct’)

severity = ‘info’

message = ‘123’

# 发布消息至交换机direct_test，且发布的消息携带的关键字routing_key是info

channel.basic_publish(exchange=’direct_test’,

routing_key=severity,

body=message)

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

connection.close()

48304ba5e6f9fe08f3fa1abda7d326ab.png

当接收端正在运行时，可以使用rabbitmqctl list_bindings来查看绑定情况。

模式3 Topic

9de6e3705659d283a44ec6006384c0e4.png

路由键模糊匹配，其实是路由键(routing_key)的扩展，就是可以使用正则表达式，和常用的正则表示式不同，这里的话“#”表示所有、全部的意思；“*”只匹配到一个词。

任何发送到Topic Exchange的消息都会被转发到所有关心routing_key中指定话题的Queue上

1.这种模式较为复杂，简单来说，就是每个队列都有其关心的主题，所有的消息都带有一个“标题”(routing_key)，Exchange会将消息转发到所有关注主题能与　　routing_key模糊匹配的队列。

2.这种模式需要routing_key，也许要提前绑定Exchange与Queue。

3.在进行绑定时，要提供一个该队列关心的主题，如“#.log.#”表示该队列关心所有涉及log的消息(一个routing_key为”MQ.log.error”的消息会被转发到该队列)。

4.“#”表示0个或若干个关键字，“*”表示一个关键字。如“log.*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.#”能与上述两者匹配。

5.同样，如果Exchange没有发现能够与routing_key匹配的Queue，则会抛弃此消息。

具体代码这里不在多余写，参照第二种模式的就可以，唯一变动的地方就是exchange type的声明，以及进行绑定和发送的时候routing_key使用正则模式即可。

至此，利用Python操作RabbitMQ，以及一些简单用法，在这里介绍完毕。

参考http://www.cnblogs.com/jishuweiwang


版权声明：本文为weixin_42160425原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。