切换到目标 vhost（如 /log）。
点击 “Exchanges” 标签。
点击 “Add a new exchange”。
填写：
- Name: log.topic.exchange
- Type: topic
- Durability: Durable（持久化，重启后保留）
- 其他选项保持默认。
点击 “Add exchange”。

⚠️ 不建议使用默认交换机（名称为空字符串），因其行为固定（Direct 类型，自动绑定所有队列），缺乏灵活性。

1.2.3 交换机与消息路由

交换机不存储消息，仅负责将消息路由到一个或多个队列。路由规则由绑定（Binding）定义，绑定包含 routing key 或 pattern。

1.3 主题（Topic）与绑定（Binding）

1.3.1 Topic 模式详解

Topic 交换机使用 routing key 的通配符匹配机制：

*：匹配一个单词（如 log.* 匹配 log.info，但不匹配 log.system.error）。
#：匹配零个或多个单词（如 log.# 匹配 log、log.info、log.system.error）。

例如：

routing key = "log.system.error" 可被 log.# 和 log.*.error 匹配。

1.3.2 在 Web 管理界面创建队列并绑定

在 “Queues” 标签下，点击 “Add a new queue”。
- Name: log.queue
- Durability: Durable
- vhost: /log
创建后，进入该队列详情页。
在 “Bindings” 区域，点击 “Bind to exchange”。
- Exchange: log.topic.exchange
- Routing key: log.#（接收所有日志）
点击 “Bind”。

此时，任何发送到 log.topic.exchange 且 routing key 符合 log.# 的消息都会进入 log.queue。

1.4 RabbitMQ Web 管理界面监控能力

1.4.1 实时状态与运行时统计

RabbitMQ 管理界面提供丰富的实时监控信息：

Overview：集群状态、节点内存、磁盘使用、连接数、通道数。
Connections / Channels：客户端连接详情。
Queues：每个队列的消息速率（publish / deliver）、消息数量（ready / unacknowledged）、消费者数量。
Exchanges：消息流入速率。

与 ActiveMQ 对比：RabbitMQ 不直接显示队列的总消息历史数量（即累计入队总数），仅显示当前队列中未消费的消息数（ready + unack）。若需统计总量，需通过监控插件（如 Prometheus + rabbitmq_exporter）或应用层记录。

1.4.2 关键监控指标

Messages ready：可被消费的消息数。
Messages unacknowledged：已投递但未确认的消息。
Publish rate：生产速率（msg/s）。
Consumer count：活跃消费者数量。

这些指标对排查积压、性能瓶颈至关重要。

2. RabbitMQ 持久化日志队列方案选型

2.1 可选方案对比

方案	交换机类型	队列持久化	消息持久化	适用场景
A	Direct	是	是	单一类型日志，如仅 error
B	Fanout	是	是	广播日志到多个存储系统
C	Topic	是	是	多级别日志（info/warn/error），灵活路由
D	Headers	是	是	基于元数据过滤，复杂但少用

2.2 最佳方案选择：Topic 交换机 + 持久化队列

选择理由：

日志通常包含多级别（info、warn、error、debug）。
Topic 模式支持通配符，便于未来扩展（如按模块 user.log.error）。
持久化确保 RabbitMQ 重启后消息不丢失。
避免使用默认交换机，提升架构清晰度。

3. 基于 Java 1.8 + RabbitMQ + MySQL 的日志持久化实现

3.1 环境准备

RabbitMQ 3.8+（启用 rabbitmq_management 插件）
MySQL 5.7+
Java 1.8

Maven 依赖：

<dependencies><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.14.2</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.28</version></dependency><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.13.3</version></dependency>
</dependencies>

3.2 RabbitMQ 资源创建（通过 Web 管理界面）

创建 vhost：/log
创建交换机：
- Name: log.topic.exchange
- Type: topic
- Durability: Durable
创建队列：
- Name: log.queue
- Durability: Durable
绑定：
- Exchange: log.topic.exchange
- Routing key: log.#

3.3 Java 生产者代码（发送日志）

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.nio.charset.StandardCharsets;
import java.util.Date;
public class LogProducer
{
private final static String EXCHANGE_NAME = "log.topic.exchange";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/log");
factory.setUsername("guest");
factory.setPassword("guest");
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
// 声明交换机（幂等操作）
channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
String[] levels = {
"info", "warn", "error"
};
for (String level : levels) {
String routingKey = "log." + level;
String message = String.format("{\"timestamp\":%d,\"level\":\"%s\",\"message\":\"Sample %s log\"}",
new Date().getTime(), level, level);
channel.basicPublish(EXCHANGE_NAME, routingKey,
MessageProperties.PERSISTENT_TEXT_PLAIN, // 消息持久化
message.getBytes(StandardCharsets.UTF_8));
System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
}
}
}
}

关键点：MessageProperties.PERSISTENT_TEXT_PLAIN 确保消息持久化。

3.4 Java 消费者代码（保存到 MySQL）

3.4.1 数据库表结构

CREATE DATABASE IF NOT EXISTS log_db;
USE log_db;
CREATE TABLE log_entries (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
timestamp BIGINT NOT NULL,
level VARCHAR(20) NOT NULL,
message TEXT NOT NULL,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

3.4.2 消费者实现

import com.rabbitmq.client.*;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.concurrent.TimeoutException;
public class LogConsumer
{
private final static String QUEUE_NAME = "log.queue";
private final static String EXCHANGE_NAME = "log.topic.exchange";
private static final String DB_URL = "jdbc:mysql://localhost:3306/log_db?useSSL=false&serverTimezone=UTC";
private static final String DB_USER = "root";
private static final String DB_PASSWORD = "password";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost("/log");
factory.setUsername("guest");
factory.setPassword("guest");
com.rabbitmq.client.Connection rabbitConn = factory.newConnection();
Channel channel = rabbitConn.createChannel();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "log.#");
System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
DeliverCallback deliverCallback = (consumerTag, delivery) ->
{
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
saveToDatabase(message);
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
// 手动ACK
};
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag ->
{
});
}
private static void saveToDatabase(String jsonMessage) {
try {
JsonNode node = new ObjectMapper().readTree(jsonMessage);
long timestamp = node.get("timestamp").asLong();
String level = node.get("level").asText();
String message = node.get("message").asText();
try (java.sql.Connection conn = DriverManager.getConnection(DB_URL, DB_USER, DB_PASSWORD);
PreparedStatement ps = conn.prepareStatement(
"INSERT INTO log_entries (timestamp, level, message) VALUES (?, ?, ?)")) {
ps.setLong(1, timestamp);
ps.setString(2, level);
ps.setString(3, message);
ps.executeUpdate();
}
} catch (Exception e) {
e.printStackTrace();
// 实际项目中应加入重试或死信队列机制
}
}
}