首頁手記 RabbitMQ資料：新手入門教程與實踐指南

RabbitMQ資料：新手入門教程與實踐指南

標簽：

中間件

概述

本文全面介绍了RabbitMQ的相关资料，包括其作用、应用场景、与其他消息队列系统的比较、安装与配置方法、核心概念以及常用操作指南。文章还提供了多个实践案例，帮助读者深入理解RabbitMQ的使用方法。

RabbitMQ简介

RabbitMQ是一种基于AMQP（高级消息队列协议）的开源消息代理软件（也被称为消息中间件）。它支持多种消息传递协议，并提供灵活的路由机制和管理界面。RabbitMQ的核心在于提供一个高度可靠的消息传递系统，使得应用程序之间可以高效地进行通信。

RabbitMQ的作用和应用场景

RabbitMQ的主要作用是作为消息的中转站，使得发送消息的应用程序（生产者）和接收消息的应用程序（消费者）能够解耦。以下是RabbitMQ的一些常见应用场景：

异步处理：将耗时的操作从主程序中移除，实现异步处理。
解耦：通过消息队列将不同的服务解耦，使得服务之间可以独立扩展和升级。
削峰填谷：在系统负载过大时，通过消息队列缓冲请求，避免系统过载。
数据分发：将数据发送给多个消费者进行处理，实现负载均衡。
日志收集：将日志信息发送到消息队列，由日志处理系统统一处理。

RabbitMQ与其他消息队列系统的比较

RabbitMQ与其他消息队列系统（如Kafka、ActiveMQ）相比，具有以下特点：

可靠性：RabbitMQ提供持久化消息存储和消息确认机制，确保消息不会丢失。
灵活性：支持多种消息路由方式，包括直接路由、主题路由、扇入路由等。
可扩展性：支持集群模式，能够在多个节点之间分发负载。
社区活跃：拥有活跃的社区支持和丰富的插件生态系统。

RabbitMQ安装与配置

RabbitMQ的安装与配置是使用该消息队列系统的前提条件。

RabbitMQ的安装步骤

在Linux上安装RabbitMQ

安装依赖包：

sudo apt-get update
sudo apt-get install erlang-nox
sudo apt-get install rabbitmq-server

启动RabbitMQ服务：
```
sudo service rabbitmq-server start
```

启动管理插件（可选，但推荐）：

sudo rabbitmq-plugins enable rabbitmq_management

访问管理界面：
打开浏览器，输入http://localhost:15672，默认用户名和密码为guest。

在Windows上安装RabbitMQ

下载RabbitMQ的Windows安装包。
解压文件并安装。

启动RabbitMQ服务：

rabbitmq-service install
rabbitmq-service enable
rabbitmq-service start

启动管理插件：

rabbitmq-plugins enable rabbitmq_management

访问管理界面：
打开浏览器，输入http://localhost:15672，默认用户名和密码为guest。

RabbitMQ的基本配置介绍

RabbitMQ可以通过配置文件或命令行参数进行配置。以下是常见的配置项：

配置文件：默认路径为/etc/rabbitmq/rabbitmq.conf。
环境变量：可以通过环境变量设置一些运行时参数。
命令行参数：启动RabbitMQ时可以通过命令行参数设置一些参数。

配置文件示例

# 设置默认用户
default_user = guest
default_pass = guest

# 设置最大连接数
maximum_connections = 1000

如何启动和停止RabbitMQ服务

启动RabbitMQ服务

Linux：
```
sudo service rabbitmq-server start
```
Windows：
```
rabbitmq-service start
```

停止RabbitMQ服务

Linux：
```
sudo service rabbitmq-server stop
```
Windows：
```
rabbitmq-service stop
```

RabbitMQ核心概念

理解RabbitMQ的核心概念是使用该系统的基础。

交换器（Exchange）

交换器是消息传递的核心组件。它根据路由键将消息路由到一个或多个队列。常见的交换器类型包括：

Direct：消息根据路由键精确匹配。
Fanout：消息广播到所有绑定的队列。
Topic：消息根据路由键模式匹配。
Headers：消息根据消息头匹配。

创建交换器的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

channel.close()
connection.close()

队列（Queue）

队列是消息存储的地方。生产者将消息发送到队列，消费者从队列中获取消息。

创建队列的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.close()
connection.close()

绑定（Binding）

绑定是交换器和队列之间的连接关系。通过绑定，消息可以从交换器路由到队列。

创建绑定的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

channel.queue_declare(queue='hello')

channel.queue_bind(exchange='direct_logs', queue='hello', routing_key='hello')

channel.close()
connection.close()

消息（Message）

消息是发送到队列的数据。消息可以包含一个路由键，用于在交换器中进行路由。

发送消息的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.basic_publish(exchange='direct_logs', routing_key='hello', body='Hello World!')

channel.close()
connection.close()

生产者（Producer）和消费者（Consumer）

生产者负责发送消息到交换器，而消费者负责从队列中获取消息并处理。

生产者示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.basic_publish(exchange='direct_logs', routing_key='hello', body='Hello World!')

channel.close()
connection.close()

消费者示例代码

import pika

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

channel.start_consuming()

RabbitMQ常用操作指南

熟悉RabbitMQ的常用操作是有效使用该系统的前提。

创建交换器、队列和绑定

创建交换器、队列和绑定是使用RabbitMQ的基础操作。

创建交换器的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

channel.close()
connection.close()

创建队列的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.close()
connection.close()

创建绑定的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

channel.queue_declare(queue='hello')

channel.queue_bind(exchange='direct_logs', queue='hello', routing_key='hello')

channel.close()
connection.close()

发送和接收消息的步骤

发送和接收消息是消息队列系统的基本功能。

发送消息的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.basic_publish(exchange='direct_logs', routing_key='hello', body='Hello World!')

channel.close()
connection.close()

接收消息的示例代码

import pika

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_consume(queue='hello', on_message_callback=callback, auto_ack=True)

channel.start_consuming()

消息确认机制（Message Acknowledgment）

消息确认机制确保消息被正确接收并处理。

发送消息并要求确认的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.basic_publish(exchange='direct_logs', routing_key='hello', body='Hello World!', properties=pika.BasicProperties(delivery_mode=2))

channel.close()
connection.close()

消费者确认消息的示例代码

import pika

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

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_consume(queue='hello', on_message_callback=callback)

channel.start_consuming()

消息持久化（Message Persistence）

消息持久化确保消息不会因服务重启而丢失。

发送持久化消息的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

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

channel.basic_publish(exchange='direct_logs', routing_key='hello', body='Hello World!', properties=pika.BasicProperties(delivery_mode=2))

channel.close()
connection.close()

创建持久化队列的示例代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

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

channel.close()
connection.close()

RabbitMQ实践案例

通过实际案例了解如何使用RabbitMQ解决实际问题。

实现简单的消息传递应用

本案例展示如何使用RabbitMQ实现简单的消息传递应用。

生产者代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_publish(exchange='', routing_key='hello', body='Hello World!')

channel.close()
connection.close()

消费者代码

import pika

def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='hello')

channel.basic_consume(queue='hello', on_message_callback=callback)

channel.start_consuming()

使用RabbitMQ构建削峰填谷的系统

本案例展示如何使用RabbitMQ构建削峰填谷的系统。

生产者代码

import pika
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

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

for i in range(20):
    message = f"Message {i}"
    channel.basic_publish(exchange='',
                          routing_key='task_queue',
                          body=message,
                          properties=pika.BasicProperties(delivery_mode=2))

    print(f" [x] Sent {message}")
    time.sleep(1)

channel.close()
connection.close()

消费者代码

import pika
import time

def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()}")
    time.sleep(body.count(b'.'))
    print(f" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

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

channel.basic_qos(prefetch_count=1)

channel.basic_consume(queue='task_queue', on_message_callback=callback)

channel.start_consuming()

模拟消息路由的场景

本案例展示如何模拟消息路由的场景。

生产者代码

import pika

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

channel.basic_publish(exchange='direct_logs', routing_key='info', body='Info Message')
channel.basic_publish(exchange='direct_logs', routing_key='warning', body='Warning Message')
channel.basic_publish(exchange='direct_logs', routing_key='error', body='Error Message')

channel.close()
connection.close()

消费者代码

import pika

def callback(ch, method, properties, body):
    print(f" [x] Received {body.decode()} from {method.routing_key}")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.exchange_declare(exchange='direct_logs', exchange_type='direct')

channel.queue_declare(queue='info_queue')
channel.queue_bind(exchange='direct_logs', queue='info_queue', routing_key='info')

channel.queue_declare(queue='warning_queue')
channel.queue_bind(exchange='direct_logs', queue='warning_queue', routing_key='warning')

channel.queue_declare(queue='error_queue')
channel.queue_bind(exchange='direct_logs', queue='error_queue', routing_key='error')

channel.basic_consume(queue='info_queue', on_message_callback=callback)
channel.basic_consume(queue='warning_queue', on_message_callback=callback)
channel.basic_consume(queue='error_queue', on_message_callback=callback)

channel.start_consuming()

點擊查看更多內容

為 TA 點贊

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

評論

評論

共同學習，寫下你的評論

評論加載中...

展開查看更多評論

作者其他優質文章

正在加載中

揚帆大魚

手記
篇

粉絲

13

獲贊與收藏

51

關注作者，訂閱最新文章

閱讀免費教程

后端通用面試教程

41個小節 32074 358

網絡編程入門教程

20個小節 13196 249

Pandas 入門教程

25個小節 19564 369

推薦

評論

收藏

共同學習，寫下你的評論



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

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

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

支付方式

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

今天注冊有機會得

100積分直接送

付費專欄免費學

大額優惠券免費領

立即參與放棄機會

點擊
抽獎

慕課手記新用戶專享福利

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

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

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

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

作者：

免費贈送

兌換碼：1111222211 復制

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

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

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

熱搜

最近搜索清空

RabbitMQ資料：新手入門教程與實踐指南

在Linux上安装RabbitMQ

在Windows上安装RabbitMQ

配置文件示例

启动RabbitMQ服务

停止RabbitMQ服务

创建交换器的示例代码

创建队列的示例代码

创建绑定的示例代码

发送消息的示例代码

生产者示例代码

消费者示例代码

创建交换器的示例代码

创建队列的示例代码

创建绑定的示例代码

发送消息的示例代码

接收消息的示例代码

发送消息并要求确认的示例代码

消费者确认消息的示例代码

发送持久化消息的示例代码

创建持久化队列的示例代码

生产者代码

消费者代码

生产者代码

消费者代码

生产者代码

消费者代码

閱讀免費教程