RabbitMQ之交换器和队列

客官°小女子只卖身不卖艺 2021-11-17 01:22 455阅读 0赞

交换器和队列是MQ中high-level层面的构建模块，应用程序需确保在使用它们的时候就已经存在了，在使用之前需要先声明它们。以下源码是基于spring-amqp-1.7.7.RELEASE.jar的源码分析

1.  交换机源码分析：

public abstract class AbstractExchange extends AbstractDeclarable implements Exchange {
        private final String name;
        private final boolean durable;
        private final boolean autoDelete;
        private final Map<String, Object> arguments;
        private volatile boolean delayed;
        private boolean internal;
    
        public AbstractExchange(String name) {
            this(name, true, false);
        }
    
        public AbstractExchange(String name, boolean durable, boolean autoDelete) {
            this(name, durable, autoDelete, (Map)null);
        }
    
        public AbstractExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
            this.name = name;
            this.durable = durable;
            this.autoDelete = autoDelete;
            if (arguments != null) {
                this.arguments = arguments;
            } else {
                this.arguments = new HashMap();
            }
        }
    }

FanoutExchange，DirectExchange和TopicExchange等交换机都继承AbstractExchange，它们交换机的声明都最终调用得是AbstractExchange的构造方法，各个参数的含义：

*  name：交换器（交换机）的名称；
 *  durable：设置是否持久化，默认值是true，表示持久化，反之非持久化。持久化可以将交换器存盘，在服务器重启的时候不会丢失相关信息；
 *  autoDelete：设置是否自动删除，默认值false，表示不自动删除。**自动删除的前提是至少有一个队列或者交换器与这个交换器绑定，之后所有与这个交换器绑定的队列或者交换器都与此解绑。不能理解为：当与此交换器连接的客户端都断开时，RabbitMQ会自动删除本交换器。**
 *  arguments：其他一些结构化参数；
 *  internal：设置是否是内置的，如果设置为true，则表示是内置的交换器，客户端程序无法直接发送消息到这个交换器中，只能通过交换器路由到交换器这种方式。

1.  队列源码分析：

public class Queue extends AbstractDeclarable {
        private final String name;
        private final boolean durable;
        private final boolean exclusive;
        private final boolean autoDelete;
        private final Map<String, Object> arguments;
    
        public Queue(String name) {
            this(name, true, false, false);
        }
    
        public Queue(String name, boolean durable) {
            this(name, durable, false, false, (Map)null);
        }
    
        public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) {
            this(name, durable, exclusive, autoDelete, (Map)null);
        }
    
        public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
            Assert.notNull(name, "'name' cannot be null");
            this.name = name;
            this.durable = durable;
            this.exclusive = exclusive;
            this.autoDelete = autoDelete;
            this.arguments = arguments;
        }
    }

队列有很多重载的构造方法，但他们底层都调用了最下面的那个，参数分析：

*  name：队列名称；
 *  durable ： 设置是否持久化，默认值为true表示持久化，持久化的队列会存盘，在服务器重启的时候可以保证不丢失相关信息；
 *  exclusive：设置是否排他，默认不排他。如果一个队列被声明为排他队列，该队列仅对首次声明它的连接可见，并在连接断开时自动删除。这里需要注意三点：1，排他队列是基于连接可见的，同一个连接的不同信道是可以同时访问同一连接创建的排他队列；2，”首次“是指如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同；3，即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除，这种队列适用于一个客户端同时发送和读取消息的应用场景。
 *  autoDelete：设置为是否自动删除，默认值false，为true则设置队列为自动删除。自动删除的前提是：至少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时，才会自动删除。
 *  arguments：设置队列的其他一些参数。

1.  队列的绑定

@Bean
        public Binding bindingSelectOrderExchange(Queue queryByOrderIdQueue, DirectExchange directExchange){
            /**
             * 绑定队列query.by到指定交换机，并指定路由key
             */
            return BindingBuilder.bind(queryByOrderIdQueue).to(directExchange).with("query.by.orderId");
        }

源码如下：

public final class BindingBuilder {
        private BindingBuilder() {
        }
    	//1，绑定队列
        public static BindingBuilder.DestinationConfigurer bind(Queue queue) {
            return new BindingBuilder.DestinationConfigurer(queue.getName(), DestinationType.QUEUE);
        }
    
    	public static final class DirectExchangeRoutingKeyConfigurer extends BindingBuilder.AbstractRoutingKeyConfigurer<DirectExchange> {
            private DirectExchangeRoutingKeyConfigurer(BindingBuilder.DestinationConfigurer destination, DirectExchange exchange) {
                super(destination, exchange.getName(), null);
            }
    		//2，指定路由key
            public Binding with(String routingKey) {
                return new Binding(this.destination.name, this.destination.type, this.exchange, routingKey, Collections.emptyMap());
            }
    
            public Binding with(Enum<?> routingKeyEnum) {
                return new Binding(this.destination.name, this.destination.type, this.exchange, routingKeyEnum.toString(), Collections.emptyMap());
            }
    
            public Binding withQueueName() {
                return new Binding(this.destination.name, this.destination.type, this.exchange, this.destination.name, Collections.emptyMap());
            }
        }
    
    	public static final class DestinationConfigurer {
    	        protected final String name;
    	        protected final DestinationType type;
    	
    	        private DestinationConfigurer(String name, DestinationType type) {
    	            this.name = name;
    	            this.type = type;
    	        }
    	
    	        public Binding to(FanoutExchange exchange) {
    	            return new Binding(this.name, this.type, exchange.getName(), "", new HashMap());
    	        }
    	
    	        public BindingBuilder.HeadersExchangeMapConfigurer to(HeadersExchange exchange) {
    	            return new BindingBuilder.HeadersExchangeMapConfigurer(this, exchange);
    	        }
    			//3，绑定到DirectExchange交换机
    	        public BindingBuilder.DirectExchangeRoutingKeyConfigurer to(DirectExchange exchange) {
    	            return new BindingBuilder.DirectExchangeRoutingKeyConfigurer(this, exchange);
    	        }
    	
    	        public BindingBuilder.TopicExchangeRoutingKeyConfigurer to(TopicExchange exchange) {
    	            return new BindingBuilder.TopicExchangeRoutingKeyConfigurer(this, exchange);
    	        }
    	
    	        public BindingBuilder.GenericExchangeRoutingKeyConfigurer to(Exchange exchange) {
    	            return new BindingBuilder.GenericExchangeRoutingKeyConfigurer(this, exchange);
    	        }
    	    }
    }