Python에서 메시지 큐(RabbitMQ)를 작동하는 방법에 대한 튜토리얼

RabbitMQ는 AMQP를 기반으로 하는 완벽하고 재사용 가능한 엔터프라이즈 메시징 시스템입니다. 이는 Mozilla Public License 오픈 소스 계약을 따릅니다. 다음 기사에서는 Python을 사용하여 RabbitMQ 메시지 큐를 작동하는 방법에 대한 튜토리얼을 주로 소개합니다.

머리말

RabbitMQ는 AMQP를 기반으로 하는 완벽하고 재사용 가능한 기업 메시징 시스템입니다. 이는 Mozilla Public License 오픈 소스 계약을 따릅니다.
MQ는 메시지 대기열(MQ)을 의미하며 애플리케이션 간 통신 방법입니다. 애플리케이션은 연결을 위한 전용 연결이 필요 없이 큐에서 메시지(애플리케이션별 데이터)를 읽고 쓰는 방식으로 통신합니다. 메시징이란 일반적으로 원격 프로시저 호출과 같은 기술에 사용되는 서로 직접 호출하는 대신 메시지로 데이터를 전송하여 서로 통신하는 프로그램을 말합니다. 큐잉은 큐를 통해 통신하는 애플리케이션을 의미합니다. 대기열을 사용하면 수신 및 송신 애플리케이션이 동시에 실행되어야 한다는 요구 사항이 제거됩니다.

애플리케이션 시나리오:

RabbitMQ는 의심할 여지 없이 현재 가장 인기 있는 메시지 대기열 중 하나이며 다양한 언어 환경에 대한 풍부한 지원도 제공합니다. .NET 개발자로서 이 도구를 배우고 이해하는 것이 필요합니다. 메시지 대기열에는 대략 세 가지 사용 시나리오가 있습니다.

1. 시스템 통합 및 분산 시스템 설계. 다양한 하위 시스템은 메시지를 통해 연결되며, 이 솔루션은 점차 '메시지를 통과하는 아키텍처'라는 아키텍처 스타일로 발전해 왔습니다.

2. 로깅 등 시스템 내 동기화 처리 방식이 처리량에 심각한 영향을 미치는 경우. 시스템에 모든 사용자 행동 로그를 기록해야 하는 경우 로그를 동기적으로 기록하면 필연적으로 시스템의 응답 속도에 영향을 미치게 됩니다. 로그 메시지를 메시지 큐에 보내면 로깅 하위 시스템이 로그 정보를 비동기적으로 사용하게 됩니다.

3. 전자상거래 깜짝 세일 시나리오 등 시스템의 고가용성. 애플리케이션 서버나 데이터베이스 서버가 특정 시간에 많은 수의 요청을 받으면 시스템 다운타임이 발생합니다. 요청이 메시지 큐로 전달될 수 있고 서버가 이러한 메시지를 소비하는 경우 요청이 더 원활해지고 시스템 가용성이 향상됩니다.

1. 설치 환경

first 설치 Linux

# 环境为CentOS 7
yum install rabbitmq-server # 安装RabbitMQ
systemctl start rabbitmq-server # 启动
systemctl enable rabbitmq-server # 开机自启
systemctl stop firewall-cmd  # 临时关闭防火墙

pip을 사용하여 Python3 개발 패키지를 설치하기 위해 소프트웨어를 설치하면 http : /에 액세스 할 수 있습니다. /115.xx.xx.xx:15672/ 내장 웹 페이지에 액세스하여 RabbitMQ를 보고 관리합니다. 기본 관리자 사용자 비밀번호는 guest

입니다. 2. 대기열에 메시지를 추가하기만 하면 됩니다

pip3 install pika

3. 대기열에서 메시지를 가져오기만 하면 됩니다

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 19:25
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Producer
import pika
# 创建连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx'))
# 创建频道对象
channel = connection.channel()
# 指定一个队列，如果该队列不存在则创建
channel.queue_declare(queue='test_queue')
# 提交消息
for i in range(10):
 channel.basic_publish(exchange='', routing_key='test_queue', body='hello,world' + str(i))
 print("sent...")
# 关闭连接
connection.close()

4. 소비자가 오프라인이 되면

다음과 같은 상황을 상상해 보세요. 소비자가 메시지 대기열에서 n개의 데이터를 가져와 처리하려고 하는데 시스템이 다운된 경우 어떻게 해야 하나요? 관리하다? RabbieMQ에는 소비자 처리 종료를 확인하는 데 사용할 수 있는 ACK가 있습니다. 이는 네트워크의 ACK와 다소 유사합니다. 소비자가 대기열에서 데이터를 얻을 때마다 대기열은 데이터를 즉시 제거하지 않고 해당 ACK를 기다립니다. 소비자는 데이터를 획득하고 처리를 완료한 후 ACK 패킷을 대기열로 보내 메시지가 처리되었으며 삭제될 수 있음을 RabbitMQ에 알립니다. 이때 RabbitMQ는 대기열에서 데이터를 제거합니다. 따라서 이 경우 소비자가 오프라인이 되더라도 문제가 없습니다. 데이터는 여전히 대기열에 존재하므로 다른 소비자가 처리할 수 있도록 남겨두게 됩니다.

은 Python에서 다음과 같이 구현됩니다.

소비자는 확인 패키지를 보내지 않는다는 의미의 코드 라인

을 가지고 있습니다. no_ack=True로 수정하면 각 처리 후에 RabbitMQ에 확인 패킷을 보내 메시지가 처리되었는지 확인합니다.

5. RabbitMQ가 다운되면 어떻게 되나요?

channel.basic_consume(callback, queue='test_queue', no_ack=False) ，其中no_ack=False

ACK 패킷이 있더라도 RabbitMQ가 다운되면 데이터는 계속 손실됩니다. 따라서 RabbitMQ에 대한 데이터 지속성 저장소를 설정할 수 있습니다. RabbitMQ는 큐가 다시 시작될 때 큐가 그대로 유지되도록 디스크에 데이터를 유지합니다.

은 Python에서 다음과 같이 구현됩니다.

我们声明一个队列是这样的channel.queue_declare(queue='test_queue') ，如果需要持久化一个队列可以这样声明channel.queue_declare(queue='test_queue', durable=True) 。不过这行直接放在代码中是不能执行的，因为以前已经有了一个名为test_queue的队列，RabbitMQ 不允许用不同的方式声明同一个队列，所以可以换一个队列名新建来指定数据持久化存储。不过如果只是这样声明的话，在 RabbitMQ 宕机重启后确实队列还在，不过队列里的数据就没有了。除非我们这样来声明队列channel.basic_publish(exchange='', routing_key="test_queue", body=message, properties=pika.BasicProperties(delivery_mode = 2,)) 。

六、最简单的发布订阅

最简单的发布订阅在 RabbitMQ 中称之为Fanout模式。也就是说订阅者订阅某个频道，然后发布者向这个频道中发布消息，所有订阅者就都能接收到这条消息。不过因为发布者需要使用订阅者创建的随机队列所以需要先启动订阅者才能启动发布者。

发布者代码：

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 20:21
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Publisher
import pika
# 创建连接对象
connection = pika.BlockingConnection(pika.ConnectionParameters(host='115.xx.xx.xx'))
# 创建频道对象
channel = connection.channel()
# 定义交换机，exchange表示交换机名称，type表示类型
channel.exchange_declare(exchange='my_fanout',
       type='fanout')
message = 'Hello Python'
# 将消息发送到交换机
channel.basic_publish(exchange='my_fanout', # 指定exchange
      routing_key='', # fanout下不需要配置，配置了也不会生效
      body=message)
connection.close()

订阅者代码：

#!/usr/bin/env python3
# coding=utf-8
# @Time : 2017/6/13 20:20
# @Author : Shawn
# @Blog : https://blog.just666.cn
# @Email : shawnbluce@gmail.com
# @purpose : RabbitMQ_Subscriber
import pika
credentials = pika.PlainCredentials('guest', 'guest')
# 连接到RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('115.xx.xx.xx', 5672, '/', credentials))
channel = connection.channel()
# 定义交换机，进行exchange声明，exchange表示交换机名称，type表示类型
channel.exchange_declare(exchange='my_fanout',
       type='fanout')
# 随机创建队列
result = channel.queue_declare(exclusive=True) # exclusive=True表示建立临时队列，当consumer关闭后，该队列就会被删除
queue_name = result.method.queue
# 将队列与exchange进行绑定
channel.queue_bind(exchange='my_fanout',
     queue=queue_name)
# 定义回调方法
def callback(ch, method, properties, body):
 print(body.decode('utf-8'))
# 从队列获取信息
channel.basic_consume(callback,
      queue=queue_name,
      no_ack=True)
channel.start_consuming()

总结

위 내용은 Python에서 메시지 큐(RabbitMQ)를 작동하는 방법에 대한 튜토리얼의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!