RabbitMQ 交换机、绑定、队列、消息、虚拟主机详解
交换机属性
交换机属性:
name: 交换机名称
type: 交换机类型 direct，topic，fanout，headers
durability: 是否须要持久化，true 为持久化
auto delete: 当最后一个绑定到 exchange 上的队列被删除后，exchange 没有绑定的队列了，自动删除该 exchange
internal: 当前 exchange 是否用于 rabbitMQ 内部使用，默认为 false
arguments: 扩展参数，用于扩展 AMQP 协议自制定化使用
复制代码
1.direct exchange类型
direct exchange: 全部发送到 direct exchange 的消息被转发到 routing key 中指定的queue
复制代码
注意:direct模式可使用 rabbitMQ 自带的 exchange:default exchange，因此不须要将 exchange 进行任何绑定(binding)操做，消息传递时，routingkey 必须彻底匹配才会被队列接收，不然该消息会被抛弃。 流转示意图以下 git
代码地址： https://github.com/hmilyos/rabbitmq-api-demo
复制代码
消费端代码：github
public class ConsumerDirectExchange {
private static final Logger log = LoggerFactory.getLogger(ConsumerDirectExchange.class);
// 声明
public final static String EXCHANGE_NAME = "test_direct_exchange";
public final static String EXCHANGE_TYPE = "direct";
public final static String QUEUE_NAME = "test_direct_queue";
public final static String ROUTING_KEY = "test.direct";
public final static String ROUTING_KEY_ERROR = "test.direct.error";
public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException,
ConsumerCancelledException, InterruptedException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
connectionFactory.setAutomaticRecoveryEnabled(true);
connectionFactory.setNetworkRecoveryInterval(3000);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
// 表示声明了一个交换机
channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE, true, false, false, null);
// 表示声明了一个队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 创建一个绑定关系:
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
// durable 是否持久化消息
QueueingConsumer consumer = new QueueingConsumer(channel);
// 参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(QUEUE_NAME, true, consumer);
while (true) {
// 获取消息，若是没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
log.info("收到消息：{}", msg);
}
}
}
复制代码
启动消费端api
上管控台查看交换机和队列是否成功建立bash
点击进去查看绑定状况服务器
生产端代码app
public class ProducerDirectExchange {
private final static Logger log = LoggerFactory.getLogger(ProducerDirectExchange.class);
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
String msg = "Hello World RabbitMQ Direct Exchange test.direct Message ... ";
log.info("生产端发送了：{}", msg);
channel.basicPublish(ConsumerDirectExchange.EXCHANGE_NAME, ConsumerDirectExchange.ROUTING_KEY, null, msg.getBytes());
// channel.basicPublish(ConsumerDirectExchange.EXCHANGE_NAME, ConsumerDirectExchange.ROUTING_KEY_ERROR, null, msg.getBytes());
channel.close();
connection.close();
}
}
复制代码
而后把生产端run一下ui
再查看消费端的日志spa
该消费端只接收 routingkey 为 test.direct 的消息，证实 direct exchange 类型的，routingkey 必须彻底匹配才会被队列接收，不然该消息会被抛弃。3d
2.topic exchange 类型
topic exchange: 全部发送到 topic exchange 的消息被转发到全部关心 routingkey 中 topic 的 queue 上 exchange 将 routingkey 和某 topic 进行模糊匹配，此时队列须要绑定一个 topic。 注意: topic 可使用通配符进行模糊匹配 # 匹配一个或多个词，注意是词 * 只能匹配一个词 例如 “log.#” 能匹配到 “log.info.oa” “log.*” 只能匹配到 “log.erro” 这种格式 具体示例图以下图，usa.news 能被 usa.#，#.news 所消费，usa.weather 能被 usa.#,#.weather 所消费...日志
代码示例： 消费端：
public class ConsumerTopicExchange {
private final static Logger log = LoggerFactory.getLogger(ConsumerTopicExchange.class);
// 声明
public static final String EXCHANGE_NAME = "test_topic_exchange";
public static final String EXCHANGE_TYPE = "topic";
public static final String QUEUE_NAME = "test_topic_queue";
public static final String ROUTING_KEY_one = "user.#";
public static final String ROUTING_KEY = "user.*";
public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException,
ConsumerCancelledException, InterruptedException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
connectionFactory.setAutomaticRecoveryEnabled(true);
connectionFactory.setNetworkRecoveryInterval(3000);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
// 1 声明交换机
channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE, true, false, false, null);
// 2 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 3 创建交换机和队列的绑定关系:
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
// durable 是否持久化消息
QueueingConsumer consumer = new QueueingConsumer(channel);
// 参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(QUEUE_NAME, true, consumer);
// 循环获取消息
while (true) {
// 获取消息，若是没有消息，这一步将会一直阻塞
Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
log.info("消费端收到消息：{}", msg);
}
}
}
复制代码
启动消费端，上管控台查看建立、绑定是否成功
确认成功后，编写生产端代码
public class ProducerTopicExchange {
private final static Logger log = LoggerFactory.getLogger(ProducerTopicExchange.class);
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
// 2 建立Connection
Connection connection = connectionFactory.newConnection();
// 3 建立Channel
Channel channel = connection.createChannel();
// 4 声明
String routingKey1 = "user.save";
String routingKey2 = "user.update";
String routingKey3 = "user.delete.abc";
String msg1 = "Hello World RabbitMQ Topic Exchange Message ..." + routingKey1;
String msg2 = "Hello World RabbitMQ Topic Exchange Message ..." + routingKey2;
String msg3 = "Hello World RabbitMQ Topic Exchange Message ..." + routingKey3;
log.info("生产端， {} ：{}", routingKey1, msg1);
channel.basicPublish(ConsumerTopicExchange.EXCHANGE_NAME, routingKey1, null, msg1.getBytes());
log.info("生产端， {} ：{}", routingKey2, msg2);
channel.basicPublish(ConsumerTopicExchange.EXCHANGE_NAME, routingKey2, null, msg2.getBytes());
log.info("生产端， {} ：{}", routingKey3, msg3);
channel.basicPublish(ConsumerTopicExchange.EXCHANGE_NAME, routingKey3, null, msg3.getBytes());
channel.close();
connection.close();
}
}
复制代码
启动生产端
消费端接收到的
routingKey3 ="user.delete.abc" 的未被接收，符合 user.* 的规则 这时候在消费端把 routingKey 修改一下， routingKey ="user.#"，重启消费端，上管控台
发现以前 * 的并无解绑，须要咱们手动解绑一下，而后再启动生产端的代码
发现三条都能接收到了，符合 # 的规则。
3.fanout exchange 类型
fanout exchange: 不处理路由键，只须要简单的将队列绑定到交换机上，发送到该交换机的消息都会被转发到于该交换机绑定的全部队列上，fanout 交换机因为不须要进行routingkey 的对比 直接发送因此绑定的 queue，因此转发消息是最快的 示意图以下图所示
代码实现：
public class ConsumerFanoutExchange {
private static final Logger log = LoggerFactory.getLogger(ConsumerFanoutExchange.class);
public static final String EXCHANGE_NAME = "test_fanout_exchange";
public static final String EXCHANGE_TYPE = "fanout";
public static final String QUEUE_NAME = "test_fanout_queue";
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
connectionFactory.setAutomaticRecoveryEnabled(true);
connectionFactory.setNetworkRecoveryInterval(3000);
Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE, true, false, false, null);
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 不设置路由键
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
QueueingConsumer consumer = new QueueingConsumer(channel);
//参数：队列名称、是否自动ACK、Consumer
channel.basicConsume(QUEUE_NAME, true, consumer);
log.info("消费端启动。。。");
//循环获取消息
while (true) {
//获取消息，若是没有消息，这一步将会一直阻塞
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String msg = new String(delivery.getBody());
log.info("消费端收到消息：{}", msg);
}
}
}
复制代码
生产端代码：
public class ProducerFanoutExchange {
private static final Logger log = LoggerFactory.getLogger(ProducerFanoutExchange.class);
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
// 2 建立Connection
Connection connection = connectionFactory.newConnection();
// 3 建立Channel
Channel channel = connection.createChannel();
// 5 发送
for (int i = 0; i < 10; i++) {
String msg = "Hello World RabbitMQ FANOUT Exchange Message ...";
log.info("生产端，routingKey{}: {}", i, msg);
channel.basicPublish(ConsumerFanoutExchange.EXCHANGE_NAME, "" + i, null, (msg + i).getBytes());
}
channel.close();
connection.close();
}
}
复制代码
先启动消费端，再启动生产端
查看消费端的日志
routingkey0-9 的都能被就收，也就至关于该交换机上全部的队列都能接收来到该交换机的消息。 headers 类型的不经常使用，就不介绍了
5.binding
binding: 绑定 exchange 和 exchange/queue 之间的链接关心。binding 中能够包含 routingkey 或者参数
6. Queue
queue: 消息队列，实际存储消息数据，durability 表示是否持久化，durable 表示是，transient 表示否。auto delete: 如选择 yes，表示当最后一个监听被移除后，该 queue 会被自动删除。
7. Message
message: 服务器和应用程序之间传送的数据 本质上就是一段数据，由 properties 和 payload(body) 组成 经常使用属性: delivery mode，headersheaders(自定义属性)，content_type,content_encoding,priority,correlation_id,reply_to,expiration,message_id,timestamp,type,user_id,app_id,cluster_id 代码实现： 消费端：
public class Consumer {
private static final Logger log = LoggerFactory.getLogger(Consumer.class);
public static void main(String[] args) throws IOException, TimeoutException,
ShutdownSignalException, ConsumerCancelledException, InterruptedException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
//2 经过链接工厂建立链接
Connection connection = connectionFactory.newConnection();
//3 经过connection建立一个Channel
Channel channel = connection.createChannel();
//4 声明（建立）一个队列
String queueName = "test001";
channel.queueDeclare(queueName, true, false, false, null);
//5 建立消费者
QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
//6 设置Channel
channel.basicConsume(queueName, true, queueingConsumer);
while (true) {
//7 获取消息
Delivery delivery = queueingConsumer.nextDelivery();
String msg = new String(delivery.getBody());
log.info("消费端: " + msg);
Map<String, Object> headers = delivery.getProperties().getHeaders();
log.info("headers get myHeaders1 value: " + headers.get("myHeaders1"));
log.info("headers get myHeaders2value: " + headers.get("myHeaders2"));
//Envelope envelope = delivery.getEnvelope();
}
}
}
复制代码
生产端：
public class Procuder {
private static final Logger log = LoggerFactory.getLogger(Procuder.class);
public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost(RabbitMQCommon.RABBITMQ_HOST);
connectionFactory.setPort(RabbitMQCommon.RABBITMQ_PORT);
connectionFactory.setVirtualHost(RabbitMQCommon.RABBITMQ_DEFAULT_VIRTUAL_HOST);
// 2 经过链接工厂建立链接
Connection connection = connectionFactory.newConnection();
// 3 经过connection建立一个Channel
Channel channel = connection.createChannel();
Map<String, Object> headers = new HashMap<>();
headers.put("myHeaders1", "111");
headers.put("myHeaders2", "222");
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(2).contentEncoding("UTF-8")
.expiration("10000").headers(headers).build();
// 4 经过Channel发送数据
for (int i = 0; i < 5; i++) {
String msg = "Hello RabbitMQ!";
// 1 exchange 2 routingKey
log.info("生产端，test001： {}", msg);
channel.basicPublish("", "test001", properties, msg.getBytes());
}
// 5 记得要关闭相关的链接
channel.close();
connection.close();
}
}
复制代码
先启动消费端，上管控台确认交换机和队列是否建立和绑定成功，再启动生产端，消费端接收到以下的信息
8. virtual host
virtual host 虚拟主机 虚拟地址，用于进行逻辑隔离，最上层的消息路由，一个 virtual host 里面能够有若干个 exchange 和 queue，可是里面不能有相同名称的 exchange 或 queue