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

為了賬號安全，請及時綁定郵箱和手機立即綁定

MQ底層原理項目實戰：新手入門教程

標簽：
架構 中間件
概述

本文深入探讨了MQ底层原理项目实战，涵盖了从消息队列的创建、配置、发送与接收消息到项目实施、测试以及性能优化等关键步骤。通过具体示例和代码，详细介绍了如何在实际项目中应用MQ，确保消息的安全传输和高效处理。MQ底层原理项目实战不仅有助于理解消息队列的工作机制，还提供了丰富的实战经验。从环境搭建到性能优化的全方位指导，内容丰富且实用。

引入MQ概念
什么是MQ

消息队列（Message Queue，简称MQ）是一种应用程序间的通信方法。消息队列的用途是解耦消息的发送者和接收者，使得发送者无需等待消息的处理完成，从而实现异步通信。消息队列通常包含一个或多个消息中间件，这些中间件负责存储和转发消息。

MQ的重要性及应用场景

消息队列在现代软件架构中扮演着重要角色，其重要性体现在以下几个方面：

  1. 异步通信：通过消息队列，发送者可以异步地传输消息，而无需等待消息的处理完成。这大大提升了系统的响应速度和吞吐量。
  2. 解耦合：消息队列使系统中的组件可以解耦，从而可以独立地开发、测试和部署不同组件，提高开发效率和系统稳定性。
  3. 可靠性和容错性：消息队列通常具有消息持久化功能，能够保证即使在网络故障或处理延迟的情况下，消息也不会丢失。
  4. 扩展性：通过消息队列，可以轻松地扩展系统，增加更多的接收者来处理消息，从而提高系统的处理能力。

应用场景

  • 日志收集和处理：将日志文件发送到消息队列，然后由专门的日志处理服务来处理。
  • 异步消息处理：例如，在电子商务网站上，订单生成后可以发送消息到消息队列，然后由后台系统处理订单详情。
  • 实时数据处理：在实时数据分析场景中，数据流通过消息队列传输，然后由实时数据处理系统进行分析。
MQ的核心原理
消息传递模型

消息传递模型是消息队列的基础，它定义了发送者与接收者之间消息的传递方式。常见的消息传递模型包括点对点模型（Point-to-Point，PTP）和发布/订阅模型（Publish/Subscribe，P/S）。

点对点模型

点对点模型下，每个消息有且仅有一个接收者。一个消息生产者（发送方）向一个队列发送消息，多个消息消费者（接收方）可以竞争访问该队列中的消息，但每次只有一个接收方能够获取一条消息并进行处理。这种模型适用于任务的分发和处理。

代码示例

下面是一个简单的点对点消息传递模型的示例，使用RabbitMQ作为消息队列：

import pika

# 创建连接到RabbitMQ服务器
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()

发布/订阅模型

发布/订阅模型适用于一对多的消息传递。在该模型下，多个消息生产者可以将消息发布到主题（Topic）上，而多个订阅者可以订阅这些主题并接收消息。这种模型适用于广播消息和事件驱动系统。

代码示例

下面是一个简单的发布/订阅消息传递模型的示例，使用RabbitMQ作为消息队列：

import pika

# 创建连接到RabbitMQ服务器
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)

# 发布消息到交换机
channel.basic_publish(exchange='logs', routing_key='', body='An info message')

# 关闭连接
connection.close()
消息队列的工作流程

消息队列的工作流程主要包括以下几个步骤：

  1. 创建消息：生产者创建并发送消息到消息队列。
  2. 存储消息：消息队列将消息存储在队列中，确保消息不丢失。
  3. 消息传递：消息队列根据相应的规则将消息传递给消费者。
  4. 处理消息：消费者从队列中获取消息并进行处理，处理完成后消息队列将删除已处理的消息。

代码示例

以下是一个简单的消息队列工作流程示例，使用RabbitMQ作为消息队列：

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='task_queue')

# 发送消息到队列
message = 'This is a task message'
channel.basic_publish(exchange='', routing_key='task_queue', body=message)
print(" [x] Sent %r" % message)

# 关闭连接
connection.close()
消息的存储和传输机制

消息队列的存储和传输机制主要分为两种：内存存储和持久化存储。

内存存储

内存存储指消息队列将消息存储在内存中，这种方式的优点是读写速度较快，但缺点是消息在系统重启后会丢失。

持久化存储

持久化存储指消息队列将消息存储到磁盘或其他持久化存储介质中，这种方式可以确保消息在系统重启后仍然存在，但读写速度相对较慢。

代码示例

以下是一个持久化存储消息的示例，使用RabbitMQ作为消息队列：

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列，设置为持久化队列
channel.queue_declare(queue='persistent_queue', durable=True)

# 发送持久化消息到队列
message = 'This is a persistent message'
channel.basic_publish(exchange='',
                      routing_key='persistent_queue',
                      body=message,
                      properties=pika.BasicProperties(
                          delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                      ))
print(" [x] Sent %r" % message)

# 关闭连接
connection.close()
MQ的常见类型
消息队列类型介绍

常见的消息队列类型包括如下几种：

  • RabbitMQ：一个开源的消息队列服务，支持多种消息传递协议，包括AMQP。
  • Apache Kafka：一个高吞吐量的分布式发布/订阅消息系统，常用于大数据处理。
  • ActiveMQ：一个基于Java的消息代理，支持多种消息协议，如JMS、STOMP等。
  • ZeroMQ：一个高性能的消息库，用于创建分布式或并行应用程序。
各类型MQ的特点对比

RabbitMQ

  • 特性：支持多种消息协议，如AMQP，具有高可用性、可扩展性、持久化等特点。
  • 应用场景：适用于多种异步消息传递场景，如任务队列、事件通知等。

Apache Kafka

  • 特性：高吞吐量、持久化、分布式、实时数据处理。
  • 应用场景：适用于大数据处理、日志收集、实时分析等。

ActiveMQ

  • 特性：支持多种消息协议，如JMS、STOMP，具有可靠的消息传递和高可用性。
  • 应用场景：适用于企业级消息传递服务，如企业应用集成、交易处理等。

ZeroMQ

  • 特性：高性能、灵活的消息库，支持多种消息传递模式。
  • 应用场景：适用于高性能分布式系统，如实时数据处理、网络通信等。
常用MQ服务举例
  • RabbitMQ：开源、支持多种消息协议，广泛应用于企业消息传递场景。
  • Apache Kafka：开源、高吞吐量、适用于大数据处理。
  • ActiveMQ：开源、支持多种消息协议，适用于企业级消息传递服务。
  • ZeroMQ：开源、高性能、适用于分布式系统的实时消息传递。

示例代码

下面是一个使用RabbitMQ创建和配置队列的示例：

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列，设置队列为持久化队列
channel.queue_declare(queue='my_queue', durable=True)

# 关闭连接
connection.close()
MQ的基础操作
创建和配置MQ

创建和配置消息队列的具体步骤取决于所使用的特定消息队列服务。

使用RabbitMQ

  1. 安装RabbitMQ：可以通过官方网站下载RabbitMQ并按照说明进行安装。
  2. 启动RabbitMQ服务：启动RabbitMQ服务后，可以在控制台中查看服务状态。
  3. 创建队列：使用RabbitMQ的API创建队列，指定队列名称和其他参数。
  4. 配置队列属性：设置队列的持久性、消息的自动删除等属性。

示例代码

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列，设置队列为持久化队列
channel.queue_declare(queue='my_queue', durable=True)

# 关闭连接
connection.close()
发送和接收消息

发送消息和接收消息是消息队列的基本操作。

发送消息

发送消息时，生产者需要连接到消息队列，将消息发送到指定的队列。

示例代码

import pika

# 创建连接到RabbitMQ服务器
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()

接收消息

接收消息时，消费者需要连接到消息队列，从队列中获取并处理消息。

示例代码

import pika

# 创建连接到RabbitMQ服务器
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)
    # 模拟消息处理时间
    import time
    time.sleep(body.count(b'.'))
    print(" [x] Done")

# 注册回调函数，设置队列消费者
channel.basic_consume(queue='hello',
                      on_message_callback=callback,
                      auto_ack=True)

# 启动消费者
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
管理消息队列

管理消息队列包括队列的创建、删除、查看队列信息等操作。

示例代码

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='my_queue')

# 获取队列信息
queue_info = channel.queue_declare(queue='my_queue', passive=True)
print("Queue name: %s" % queue_info.method.queue)
print("Messages in use: %d" % queue_info.method.message_count)

# 删除队列
channel.queue_delete(queue='my_queue')

# 关闭连接
connection.close()
MQ项目实战案例
项目选型及需求分析

项目选型是指根据项目的具体需求选择合适的消息队列服务。需求分析包括以下几个方面：

  1. 系统架构：项目是否需要异步处理、解耦合、高可用性等特性。
  2. 消息传递模式：项目需要使用点对点模型还是发布/订阅模型。
  3. 性能要求：项目需要处理的吞吐量和延迟要求。
  4. 消息持久性：项目是否需要持久化消息存储，以便在系统重启后仍然保留消息。

示例

假设我们正在为一个电子商务平台开发订单处理系统，该系统需要实现以下功能：

  • 异步处理订单：订单生成后，需要异步地处理订单详情，如支付、配送等。
  • 解耦合：订单生成模块和订单处理模块需要解耦，以便独立开发和部署。
  • 消息持久化：需要确保订单消息在系统重启后仍然存在，并不会丢失。

根据以上需求，可以选择使用RabbitMQ作为消息队列服务，因为它支持多种消息传递模式，并且具有持久化消息存储功能。

示例代码

发送端实现

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='order_processing', durable=True)

# 发送持久化消息到队列
message = 'This is an order message'
channel.basic_publish(exchange='',
                      routing_key='order_processing',
                      body=message,
                      properties=pika.BasicProperties(
                          delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                      ))
print(" [x] Sent %r" % message)

# 关闭连接
connection.close()

接收端实现

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='order_processing', durable=True)

# 定义回调函数，处理接收到的消息
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    # 模拟订单处理时间
    import time
    time.sleep(body.count(b'.'))
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 注册回调函数，设置队列消费者
channel.basic_consume(queue='order_processing',
                      on_message_callback=callback,
                      auto_ack=False)

# 启动消费者
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
项目实施步骤

项目实施包括以下几个步骤：

  1. 环境搭建：安装并配置RabbitMQ服务。
  2. 发送端实现：实现订单生成模块，将订单消息发送到消息队列。
  3. 接收端实现：实现订单处理模块，从消息队列中获取订单消息并进行处理。
  4. 消息持久化：确保消息队列中的消息在系统重启后仍然存在。

示例代码

发送端实现

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='order_processing', durable=True)

# 发送持久化消息到队列
message = 'This is an order message'
channel.basic_publish(exchange='',
                      routing_key='order_processing',
                      body=message,
                      properties=pika.BasicProperties(
                          delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                      ))
print(" [x] Sent %r" % message)

# 关闭连接
connection.close()

接收端实现

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='order_processing', durable=True)

# 定义回调函数，处理接收到的消息
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    # 模拟订单处理时间
    import time
    time.sleep(body.count(b'.'))
    print(" [x] Done")
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 注册回调函数，设置队列消费者
channel.basic_consume(queue='order_processing',
                      on_message_callback=callback,
                      auto_ack=False)

# 启动消费者
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
项目测试与优化

项目测试包括以下几个方面：

  1. 功能测试：确保消息发送和接收功能正常。
  2. 性能测试：测试消息队列的吞吐量和延迟。
  3. 容错测试：测试系统在消息丢失、网络故障等情况下的表现。

示例代码

功能测试

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='test_queue')

# 发送消息到队列
message = 'This is a test message'
channel.basic_publish(exchange='',
                      routing_key='test_queue',
                      body=message)
print(" [x] Sent %r" % message)

# 关闭连接
connection.close()

性能测试

性能测试可以通过发送大量的消息并测量处理时间来实现。可以使用timeit库来测量发送和接收消息的性能。

import timeit
import pika

def send_message():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue')
    channel.basic_publish(exchange='',
                          routing_key='test_queue',
                          body='Test message')
    connection.close()

def receive_message():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue')

    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)
    channel.start_consuming()

send_time = timeit.timeit(send_message, number=1000)
receive_time = timeit.timeit(receive_message, number=1000)
print("Send time: %f seconds" % send_time)
print("Receive time: %f seconds" % receive_time)

容错测试

容错测试可以通过模拟网络故障、消息丢失等情况来实现。可以使用pika库中的异常处理机制来测试系统的容错性。

import pika

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

try:
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue')
    channel.basic_consume(queue='test_queue', on_message_callback=on_message)
    channel.start_consuming()
except pika.exceptions.ConnectionClosedByBroker:
    print("Connection closed by broker")
except pika.exceptions.AMQPConnectionError:
    print("Connection error")
finally:
    connection.close()
MQ常见问题与解决方案
常见错误排查

在使用消息队列时，可能会遇到一些常见的错误。以下是一些常见的错误及其解决方案：

错误一：无法连接到消息队列服务器

错误信息：连接到消息队列服务器时出现错误。

解决方案：检查消息队列服务器的网络配置，确保服务器地址和端口正确。也可以尝试重启消息队列服务来解决连接问题。

示例代码

import pika

try:
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='test_queue')
    print("Connected to RabbitMQ server")
    connection.close()
except pika.exceptions.ConnectionClosedByBroker:
    print("Connection closed by broker")
except pika.exceptions.AMQPConnectionError:
    print("Connection error")

错误二：消息丢失

错误信息：消息在发送或接收过程中丢失。

解决方案：检查消息队列的持久化配置，确保消息在发送时设置了持久化属性。另外，确保接收端正确地处理了消息，避免消息被删除而未处理。

示例代码

import pika

# 发送持久化消息到队列
message = 'This is a persistent message'
channel.basic_publish(exchange='',
                      routing_key='persistent_queue',
                      body=message,
                      properties=pika.BasicProperties(
                          delivery_mode=pika.spec.PERSISTENT_DELIVERY_MODE
                      ))
print(" [x] Sent %r" % message)

错误三：消息延迟

错误信息：消息在接收过程中延迟，导致处理时间过长。

解决方案：优化消息接收端的处理逻辑，减少消息处理时间。也可以增加接收端的处理能力，如增加更多的消费者来处理消息。

示例代码

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='test_queue')

# 定义回调函数，处理接收到的消息
def callback(ch, method, properties, body):
    print(" [x] Received %r" % body)
    # 模拟消息处理时间
    import time
    time.sleep(body.count(b'.'))
    print(" [x] Done")
    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()
性能优化策略

性能优化主要包括以下几个方面：

  1. 增加接收者：增加更多的消息消费者来处理消息，提高处理能力。
  2. 减少处理时间：优化消息处理逻辑，减少消息处理时间。
  3. 消息分片：将大消息拆分成小消息，减少单个消息的处理时间。
  4. 批量处理：使用批量处理机制，减少消息传递的次数。

示例代码

增加接收者

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='test_queue')

# 定义回调函数，处理接收到的消息
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)

# 启动多个消费者
for i in range(5):
    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()

批量处理

import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='test_queue')

# 定义回调函数，处理接收到的消息
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)

# 批量发送消息
for i in range(1000):
    channel.basic_publish(exchange='',
                          routing_key='test_queue',
                          body='Test message %d' % i)

print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()
安全性考虑

安全性是消息队列的重要方面，需要确保消息的传输和存储安全。

措施一：加密通信

保证消息传输的安全性，可以使用SSL/TLS加密通信。例如，RabbitMQ支持通过SSL/TLS协议来进行加密通信。

示例代码

import pika

# 创建连接到RabbitMQ服务器，使用SSL/TLS加密
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost',
                                                               port=5671,
                                                               credentials=pika.PlainCredentials('guest', 'guest'),
                                                               ssl_options={
                                                                   'ca_certs': '/path/to/ca_certs.pem',
                                                                   'certfile': '/path/to/certfile.pem',
                                                                   'keyfile': '/path/to/keyfile.pem',
                                                                   'cert_reqs': pika.ssl_protocol_constants.CERT_REQUIRED,
                                                               }))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='secure_queue')

# 发送消息到队列
message = 'This is a secure message'
channel.basic_publish(exchange='',
                      routing_key='secure_queue',
                      body=message)

# 关闭连接
connection.close()

措施二：消息签名

为了确保消息的完整性，可以使用消息签名机制。例如，使用HMAC算法对消息进行签名，并在接收端验证签名。

示例代码

import pika
import hmac
import hashlib

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 创建队列
channel.queue_declare(queue='signed_queue')

# 发送消息到队列，附加签名
message = 'This is a signed message'
message_signature = hmac.new(b'secret_key', message.encode(), hashlib.sha256).hexdigest()
channel.basic_publish(exchange='',
                      routing_key='signed_queue',
                      body=message + ' ' + message_signature)

# 关闭连接
connection.close()

# 接收端验证签名
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 定义回调函数，处理接收到的消息
def callback(ch, method, properties, body):
    message, signature = body.split(' ')
    if hmac.new(b'secret_key', message.encode(), hashlib.sha256).hexdigest() == signature:
        print(" [x] Received %r" % message)
    else:
        print(" [x] Invalid signature")

# 注册回调函数，设置队列消费者
channel.basic_consume(queue='signed_queue',
                      on_message_callback=callback,
                      auto_ack=True)

# 启动消费者
print(' [*] Waiting for messages. To exit press CTRL+C')
channel.start_consuming()

措施三：访问控制

确保只有授权的用户才能访问消息队列。可以使用RabbitMQ的访问控制机制来实现，例如，配置用户权限和队列权限。

示例代码


import pika

# 创建连接到RabbitMQ服务器
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost',
                                                               credentials=pika.PlainCredentials('user', 'password')))
channel = connection.channel()

# 创建队列，并设置权限
channel.queue_declare(queue='controlled_queue')
channel.queue_bind(exchange='amq.direct',
                   queue='controlled_queue',
                   routing_key='controlled_queue',
                   arguments={'x-product': 'custom_product'})
channel.queue_bind(exchange='amq.direct',
                   queue='controlled_queue',
                   routing_key='controlled_queue',
                   arguments={'x-permission': 'read-only'})

# 发送消息到队列
message = 'This is a controlled message'
channel.basic_publish(exchange='',
                      routing_key='controlled_queue',
                      body=message)

# 关闭连接
connection.close()
``

以上是MQ底层原理项目实战的总结，包括消息队列的创建、配置、发送和接收消息、项目实施和测试、性能优化和安全性考虑。希望本文能够帮助读者更好地理解和使用消息队列技术。
點擊查看更多內容
為 TA 點贊

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

評論

作者其他優質文章

正在加載中
  • 推薦
  • 評論
  • 收藏
  • 共同學習，寫下你的評論
感謝您的支持，我會繼續努力的～
掃碼打賞，你說多少就多少
贊賞金額會直接到老師賬戶
支付方式
打開微信掃一掃，即可進行掃碼打賞哦
今天注冊有機會得

100積分直接送

付費專欄免費學

大額優惠券免費領

立即參與 放棄機會
微信客服

購課補貼
聯系客服咨詢優惠詳情

 幫助反饋 APP下載

慕課網APP
您的移動學習伙伴

 公眾號

掃描二維碼
關注慕課網微信公眾號

舉報

0/150
提交
取消