RabbitMQ核心概念深度解析
引言
RabbitMQ是开源的消息代理软件，实现了高级消息队列协议（AMQP）。它提供可靠的消息传递、灵活的路由和集群部署能力。
核心概念
1.1 生产者、消费者、Broker
┌─────────┐ ┌─────────┐ ┌─────────┐
│Producer│────▶│ Broker │────▶│Consumer│
└─────────┘ └─────────┘ └─────────┘
│
▼
┌───────────┐
│ Exchange │
└───────────┘
│
┌───────────┼───────────┐
▼ ▼ ▼
┌───────┐ ┌───────┐ ┌───────┐
│Queue 1│ │Queue 2│ │Queue 3│
└───────┘ └───────┘ └───────┘
1.2 核心组件
import com.rabbitmq.client.*;
/**
* RabbitMQ基础连接
*/
public class RabbitMQBasics {
private ConnectionFactory factory;
private Connection connection;
private Channel channel;
/**
* 创建连接
*/
public void createConnection() throws Exception {
factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
connection = factory.newConnection();
channel = connection.createChannel();
}
/**
* 创建队列
*/
public void createQueue() throws Exception {
channel.queueDeclare("my-queue", true, false, false, null);
}
/**
* 发送消息
*/
public void sendMessage(String message) throws Exception {
channel.basicPublish("", "my-queue",
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes());
}
/**
* 接收消息
*/
public void consumeMessage() throws Exception {
channel.basicConsume("my-queue", true,
new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body) {
String message = new String(body);
System.out.println("Received: " + message);
}
});
}
}
消息属性
2.1 消息属性
/**
* 消息属性配置
*/
public class MessageProperties {
/**
* 创建消息属性
*/
public static AMQP.BasicProperties createProperties() {
return new AMQP.BasicProperties.Builder()
.contentType("application/json")
.deliveryMode(2) // 持久化
.priority(5)
.expiration("10000") // 过期时间
.messageId("msg-001")
.timestamp(new Date())
.build();
}
}
总结
RabbitMQ的核心概念包括生产者、消费者、Broker、交换机和队列等，理解这些概念是使用RabbitMQ的基础。
