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 必须完全匹配才会被队列接收，否则该消息会被抛弃。 流转示意图如下

代码地址：    https://github.com/hmilyos/rabbitmq-api-demo
复制代码

消费端代码：

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);
		}
	}
}
复制代码

启动消费端

上管控台查看交换机和队列是否成功创建

点击进去查看绑定情况

生产端代码

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());
		
	}
	
}
复制代码

然后把生产端run一下

再查看消费端的日志

该消费端只接收 routingkey 为 test.direct 的消息，证明 direct exchange 类型的，routingkey 必须完全匹配才会被队列接收，否则该消息会被抛弃。

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