首頁手記 RabbitMQ入門教程：輕松掌握消息隊列

RabbitMQ入門教程：輕松掌握消息隊列

標簽：

中間件

RabbitMQ简介

RabbitMQ是什么

RabbitMQ 是由Erlang编程语言开发的一个开源消息代理实现，它实现了高级消息队列协议（AMQP）。作为纯软件实现的消息代理，RabbitMQ支持多种消息传递协议，包括AMQP和MQTT。它支持多种编程语言，如Java、Python、Ruby、C#、C/C++、Erlang等。

RabbitMQ的作用和应用场景

RabbitMQ作为消息传递系统，提供了在分布式系统中可靠传递消息的能力，适用于多种场景：

异步通信：在分布式系统中，不同的组件可以通过消息队列实现异步通信。
解耦组件：将系统中的不同部分解耦，确保一个组件的变更不会影响其他组件。
任务队列：用于处理大量任务的队列，例如邮件、图片处理等。
数据存储：将数据暂存于队列中，再进行处理或转发。
负载均衡：将消息分发给多个消费者，实现负载均衡。

RabbitMQ与消息队列的关系

RabbitMQ是一个消息代理，负责在不同的客户端之间传递消息。消息队列是RabbitMQ中的核心概念，用于存储和转发消息。生产者将消息发送到队列中，消费者从队列中接收消息。RabbitMQ使用交换机和绑定来路由消息到正确的队列中。

安装与配置RabbitMQ

安装RabbitMQ

安装RabbitMQ的步骤如下：

安装Erlang模块：RabbitMQ依赖于Erlang模块，因此首先需要安装Erlang。

# Ubuntu/Debian
sudo apt-get update
sudo apt-get install erlang

# CentOS/RHEL
sudo yum install erlang

安装RabbitMQ：使用包管理器安装RabbitMQ。

# Ubuntu/Debian
sudo apt-get install rabbitmq-server

# CentOS/RHEL
sudo yum install rabbitmq-server

启动RabbitMQ服务：

sudo systemctl start rabbitmq-server
sudo systemctl enable rabbitmq-server

配置RabbitMQ

RabbitMQ的配置可以通过配置文件或命令行工具完成。以下是常见的配置步骤：

查看RabbitMQ配置：使用命令查看RabbitMQ当前配置。
```
sudo rabbitmqctl status
```

添加用户和权限：创建一个新的用户并配置权限。

sudo rabbitmqctl add_user myuser mypassword
sudo rabbitmqctl set_user_tags myuser administrator
sudo rabbitmqctl set_permissions -p / myuser ".*" ".*" ".*"

设置默认虚拟主机：RabbitMQ使用虚拟主机（Vhost）来隔离不同的消息队列。

sudo rabbitmqctl add_vhost myvhost
sudo rabbitmqctl set_permissions -p myvhost myuser ".*" ".*" ".*"

启用管理插件：管理插件提供了一个Web界面来监控和管理RabbitMQ。
```
sudo rabbitmq-plugins enable rabbitmq_management
```
访问管理界面：默认情况下，管理界面可以通过http://localhost:15672访问，使用默认用户名密码guest:guest登录。

RabbitMQ核心概念

概念介绍：节点、队列、交换机、绑定

节点：在RabbitMQ中，节点是指运行RabbitMQ的服务器实例，可以是单个节点或多个节点组成的集群。
队列：队列是消息的存储容器，生产者将消息发送到队列中，消费者从队列中接收消息。队列有两种类型：临时队列和持久化队列，其中持久化队列即使在重启后仍然存在。
交换机：交换机是消息传递系统中的路由组件，接收消息并将其路由到一个或多个队列中。常见交换机类型包括fanout（广播）、direct（定向）、topic（主题）和headers（头信息）。
绑定：绑定将交换机与队列关联，定义了交换机如何将消息路由到队列中的规则。

消息的生产和消费流程

消息的生产和消费流程包括以下几个步骤：

生产者：生产者创建并发送消息到交换机。
交换机：交换机根据绑定规则将消息路由到一个或多个队列。
队列：消息存储在队列中，等待被消费者消费。
消费者：消费者从队列中接收消息并处理。

以下是使用Python编写的简单生产者和消费者代码示例：

生产者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    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()

if __name__ == '__main__':
    main()

消费者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='hello')
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

RabbitMQ常用消息模式

简单模式(Simple)

简单模式是最基本的消息传递模式，生产者将消息发送到队列中，消费者从队列中接收消息，适用于简单的消息传递场景。

生产者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='simple_queue')
    channel.basic_publish(exchange='', routing_key='simple_queue', body='Simple message')
    print(" [x] Sent 'Simple message'")
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='simple_queue')
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    channel.basic_consume(queue='simple_queue', on_message_callback=callback, auto_ack=True)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

工作队列模式(Work Queues)

工作队列模式是将任务发送到队列中，多个消费者可以同时从队列中获取任务并处理。

生产者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='work_queue')
    for i in range(10):
        message = "Task %d" % i
        channel.basic_publish(exchange='', routing_key='work_queue', body=message)
        print(" [x] Sent %r" % message)
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例

import pika
import time

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='work_queue')
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
        time.sleep(body.count(b'.'))
        print(" [x] Done")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    channel.basic_consume(queue='work_queue', on_message_callback=callback, auto_ack=False)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

发布/订阅模式(Publish/Subscribe)

发布/订阅模式允许多个生产者将消息发布到一个主题，多个消费者可以订阅这个主题来接收消息，适用于一对多的消息传递场景。

生产者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='fanout')
    message = 'Message to all subscribers'
    channel.basic_publish(exchange='logs', routing_key='', body=message)
    print(" [x] Sent %r" % message)
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='fanout')
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    channel.queue_bind(exchange='logs', queue=queue_name)
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

路由模式(Routing)

路由模式允许多个生产者将消息发送到路由键，多个消费者可以通过绑定不同的路由键来接收消息，适用于多对多的消息传递场景。

生产者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='direct')
    severity = 'info'
    message = 'Message with severity %s' % severity
    channel.basic_publish(exchange='logs', routing_key=severity, body=message)
    print(" [x] Sent %r" % message)
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='direct')
    queue_name, _ = channel.queue_declare(queue='')
    channel.queue_bind(exchange='logs', queue=queue_name.method.queue, routing_key='info')
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    channel.basic_consume(queue=queue_name.method.queue, on_message_callback=callback, auto_ack=True)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

通配符模式(Topic)

通配符模式允许多个生产者将消息发送到通配符路由键，多个消费者可以通过绑定不同的通配符路由键来接收消息，适用于灵活的消息传递场景。

生产者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='topic')
    routing_key = 'user.info'
    message = 'Message with routing key %s' % routing_key
    channel.basic_publish(exchange='logs', routing_key=routing_key, body=message)
    print(" [x] Sent %r" % message)
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.exchange_declare(exchange='logs', exchange_type='topic')
    queue_name, _ = channel.queue_declare(queue='')
    channel.queue_bind(exchange='logs', queue=queue_name.method.queue, routing_key='user.*')
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
    channel.basic_consume(queue=queue_name.method.queue, on_message_callback=callback, auto_ack=True)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

实战演练

编写简单的生产者和消费者代码

在实战演练部分，将使用Python编写简单的生产者和消费者代码，并介绍消息确认机制和持久化消息。

生产者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue', durable=True)
    message = 'Persistent message'
    channel.basic_publish(exchange='',
                          routing_key='test_queue',
                          body=message,
                          properties=pika.BasicProperties(
                              delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                          ))
    print(" [x] Sent %r" % message)
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue', durable=True)
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    channel.basic_consume(queue='test_queue', on_message_callback=callback, auto_ack=False)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

消息确认机制

消息确认机制确保消息被成功传递并处理，消费者在处理完消息后会通过basic_ack方法向生产者确认消息已成功接收，如果消费者在处理消息过程中崩溃，消息将重新发送到队列中。

生产者代码示例（带确认机制）

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue', durable=True)
    message = 'Message with ack'
    channel.basic_publish(exchange='',
                          routing_key='test_queue',
                          body=message,
                          properties=pika.BasicProperties(
                              delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                          ))
    print(" [x] Sent %r" % message)
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例（带确认机制）

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue', durable=True)
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
        print(" [x] Processing...")
        time.sleep(2)
        print(" [x] Acknowledging...")
        ch.basic_ack(delivery_tag=method.delivery_tag)
    channel.basic_consume(queue='test_queue', on_message_callback=callback, auto_ack=False)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

持久化消息

持久化消息确保即使在RabbitMQ重启后，消息仍然存在，通过设置delivery_mode为2来实现。

生产者代码示例（持久化消息）

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue', durable=True)
    message = 'Persistent message'
    channel.basic_publish(exchange='',
                          routing_key='test_queue',
                          body=message,
                          properties=pika.BasicProperties(
                              delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                          ))
    print(" [x] Sent %r" % message)
    connection.close()

if __name__ == '__main__':
    main()

消费者代码示例（持久化消息）

import pika

def main():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue', durable=True)
    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    channel.basic_consume(queue='test_queue', on_message_callback=callback, auto_ack=False)
    print(' [*] Waiting for messages. To exit press CTRL+C')
    channel.start_consuming()

if __name__ == '__main__':
    main()

常见问题及解决方法

常见错误及解决方案

连接失败：确保RabbitMQ服务正在运行，检查网络连接和防火墙设置。
权限问题：检查用户权限是否正确，确保用户有足够的权限访问队列和交换机。
队列不存在：确保队列已经创建，或使用自动声明队列的方式。
消息丢失：检查消息是否已设置为持久化，确保队列和消息都设置为持久化。
消息未被消费：检查消费者是否正确配置，并确保消息被正确路由到队列中。

性能优化技巧

使用持久化消息：持久化消息可以确保消息在RabbitMQ重启后仍然存在。
使用批量发布：批量发送消息可以减少网络开销，提高性能。
调整队列和交换机的参数：根据实际需求调整队列和交换机的参数，如队列的预取值、交换机的类型等。
使用高可用集群：部署RabbitMQ集群可以提高系统的可用性和性能。
监控和日志：定期监控RabbitMQ的运行状态和日志，及时发现和解决问题。

通过以上内容，您已经掌握了RabbitMQ的基本概念和使用方法，可以开始尝试在实际项目中使用RabbitMQ进行消息传递了。如果有任何问题，可以参考官方文档或社区资源进行进一步的学习和参考。

點擊查看更多內容

為 TA 點贊

若覺得本文不錯，就分享一下吧！

評論

評論

共同學習，寫下你的評論

評論加載中...

展開查看更多評論

作者其他優質文章

正在加載中

一只萌萌小番薯

手記
篇

粉絲

11

獲贊與收藏

55

關注作者，訂閱最新文章

閱讀免費教程

后端通用面試教程

41個小節 32074 358

網絡編程入門教程

20個小節 13196 249

Pandas 入門教程

25個小節 19564 369

推薦

評論

收藏

共同學習，寫下你的評論



感謝您的支持，我會繼續努力的～

掃碼打賞，你說多少就多少

贊賞金額會直接到老師賬戶

支付方式

打開微信掃一掃，即可進行掃碼打賞哦

今天注冊有機會得

100積分直接送

付費專欄免費學

大額優惠券免費領

立即參與放棄機會

點擊
抽獎

慕課手記新用戶專享福利

恭喜你，你的運氣太好了，居然抽中了 100個積分！

恭喜你，抽中了價值元的專欄！

太棒了，直接落到你賬戶里！

積分商城里的羅技鼠標、機械鍵盤、
Kindle 閱讀器、小米平衡車
Apple iPad （10.2英寸）、大額優惠券
在等著你去兌換了噢

作者：

免費贈送

兌換碼：1111222211 復制

優惠券可用于購買實戰課、體系課
無門檻使用

先去看看，有什么好東西馬上兌換我愛學習，選課去


亚洲在线久爱草,狠狠天天香蕉网,天天搞日日干久草,伊人亚洲日本欧美

熱搜

最近搜索清空

RabbitMQ入門教程：輕松掌握消息隊列

RabbitMQ是什么

RabbitMQ的作用和应用场景

RabbitMQ与消息队列的关系

安装RabbitMQ

配置RabbitMQ

概念介绍：节点、队列、交换机、绑定

消息的生产和消费流程

生产者代码示例

消费者代码示例

简单模式(Simple)

生产者代码示例

消费者代码示例

工作队列模式(Work Queues)

生产者代码示例

消费者代码示例

发布/订阅模式(Publish/Subscribe)

生产者代码示例

消费者代码示例

路由模式(Routing)

生产者代码示例

消费者代码示例

通配符模式(Topic)

生产者代码示例

消费者代码示例

编写简单的生产者和消费者代码

生产者代码示例

消费者代码示例

消息确认机制

生产者代码示例（带确认机制）

消费者代码示例（带确认机制）

持久化消息

生产者代码示例（持久化消息）

消费者代码示例（持久化消息）

常见错误及解决方案

性能优化技巧

閱讀免費教程