amqp 和 exchange 详细解释

时间:2022-09-25 20:33:54

amqp  的 excange 字面意思是一个交换机。他的任务是吧 消息 分配给消息队列。

amqp 的  exchange 有三种,分别是 Direct , fanout 和 toppic。三种。

  Direct:通过 Routing key 来分配消息 应该分配给那个消息队列。在给交换机绑定 消息对列的时候需要指定  路由关键字,并且之歌路由关键字必须是不包含通配符。

      特点:消息明确,只有一个对列会消费这个消息

      官方解释:转发消息到routingKey中指定的队列

            要求队列绑定时使用的bindingKey和发送时使用routingKey的保持一致,保证只有key匹配的队列中才可以进行收发消息

 

 

  fanout:把消息分给这个 交换机下面的所有 消息队列,值得注意的是 fanout 类型的 绑定 消息对列的时候不需要指配  Routing key

      特点:分配给全部的绑定在这个交换机上的消息队列。类似于发布订阅机制。

      官方解释:

        转发消息到与该交换机绑定的所有队列

        只要接收端和发送端使用同一个交换机,所有端都可以收发消息

 

 

  toppic: 上面理想个的综合,把消息分配绑定在这个交换机上的多个消息队列,但是 不一定是全部。可能一个也没有,可能全部都有。通过  带有通配符的 路由关键在来指定分配规则

      个 在 绑定 交换机 和queue 关系的时候 ,Routing key  配置成带有 通配符的 。发消息的 时候 发一个明确的消息 Routing key ,这样 这个消息就会分配到 合适的 消息队列中了。

      特点:分配给 多个 消息队列。可以灵活指定。

      官方解释:

        转发消息到所有关心routingkey中指定话题的队列,只要队列关心的主题(bindingkey)能与消息带的routingkey模糊匹配,就可以将消息发送到该队列。队列绑定时提供的主题可以使用"*"和"#"来的表示关键字,"*"表示一个关键字,

        "#"代表0个或若干个关键  字。关键字之间用"."分隔,如:有routingkey:"log","log.out","log.a.bug"; bindingKey为"log.*"的队列只能接收"log.out"的消息,而bindingKey为"log.#"的队列可以接收前面三个消息。

 

 

   header:(我也没用过 猜的)

      

      有时消息的路由操作会涉及到多个属性,此时使用消息头就比用路由键更容易表达,头交换机(headers exchange)就是为此而生的。头交换机使用多个消息属性来代替路由键建立路由规则。

      通过判断消息头的值能否与指定的绑定相匹配来确立路由规则。

      我们可以绑定一个队列到头交换机上,并给他们之间的绑定使用多个用于匹配的头(header)。这个案例中,消息代理得从应用开发者那儿取到更多一段信息,换句话说,它需要考虑某条消息(message)

      是需要部分匹配还是全部匹配。上边说的“更多一段消息”就是"x-match"参数。当"x-match"设置为“any”时,消息头的任意一个值被匹配就可以满足条件,而当"x-match"设置为“all”的时候,就需要消息头的所有值都匹配成功。

      头交换机可以视为直连交换机的另一种表现形式。头交换机能够像直连交换机一样工作,不同之处在于头交换机的路由规则是建立在头属性值之上,而不是路由键。路由键必须是一个字符串,而头属性值则没有这个约束,

      它们甚至可以是整数或者哈希值(字典)等。

      BindingBuilder.bind( queueWeixinPublicDL ).to( new HeadersExchange("") ).whereAny(new HashMap<>());

     

       解释:如果 消息中有属性匹配 header 那么就关系就成立 。

 

关于  amqp  消息的 传递流程:

  生产者连接到消息服务器(broker)  --> 生产者发送消息到 交换机( exchange) --> 交换机 把消息路由给绑定在其上的消息队列(queue)  -->  消费者连接到 消息服务器 从 消息对列中取出消息- -> 消费并且告诉 消息队列这个消息已经正常消费(ack)

 

 

一些 amqp 的 名词解释:

  • Broker: 接收和分发消息的应用,RabbitMQ Server就是Message Broker。
  • Virtual host: 出于多租户和安全因素设计的,把AMQP的基本组件划分到一个虚拟的分组中,类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时,可以划分出多个vhost,每个用户在自己的vhost创建exchange/queue等。
  • Connection: publisher/consumer和broker之间的TCP连接。断开连接的操作只会在client端进行,Broker不会断开连接,除非出现网络故障或broker服务出现问题。
  • Channel: 如果每一次访问RabbitMQ都建立一个Connection,在消息量大的时候建立TCP Connection的开销将是巨大的,效率也较低。Channel是在connection内部建立的逻辑连接,如果应用程序支持多线程,通常每个thread创建单独的channel进行通讯,AMQP method包含了channel id帮助客户端和message broker识别channel,所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。
  • Exchange: message到达broker的第一站,根据分发规则,匹配查询表中的routing key,分发消息到queue中去。常用的类型有:direct (point-to-point), topic (publish-subscribe) and fanout (multicast)。
  • Queue: 消息最终被送到这里等待consumer取走。一个message可以被同时拷贝到多个queue中。
  • Binding: exchange和queue之间的虚拟连接,binding中可以包含routing key。Binding信息被保存到exchange中的查询表中,用于message的分发依据。

 

 下面的例子都是 基于rabbit mq 的

传统xml 方式的  配置方式:

<!-- 容邦消息队列 -->	
	<rabbit:queue name="rbNotificationEventQueue" durable="true"
		auto-delete="false" exclusive="false" />
	<!-- 容邦消息死信队列 -->	
	<rabbit:queue name="rbNotificationEventQueueDL" durable="true" auto-delete="false" exclusive="false">
	    <rabbit:queue-arguments>
	        <entry key="x-dead-letter-exchange" value="rbNotificationEventExchange"/>
	    	<entry key="x-dead-letter-routing-key" value="rbNotificationEventQueue"/>
	    </rabbit:queue-arguments>
    </rabbit:queue>	
		

	<!-- 容邦通知查询事件交换机 -->
	<rabbit:direct-exchange name="rbNotificationEventExchange"
		durable="true" auto-delete="false">
		<rabbit:bindings>
			<rabbit:binding queue="rbNotificationEventQueue" key="rbNotificationEventQueue" />
			<rabbit:binding queue="rbNotificationEventQueueDL" key="rbNotificationEventQueueDL" />
		</rabbit:bindings>
	</rabbit:direct-exchange>
	
	<!--监听配置 -->
	<rabbit:listener-container
		connection-factory="connectionFactory" acknowledge="manual">
		<rabbit:listener queues="rbNotificationEventQueue" ref="queueListenter" />
	</rabbit:listener-container>

  

解释:定义了一个 直连交换机(rbNotificationEventExchange) ,上面绑定了两个消息队列。rbNotificationEventQueue 是一个正常队列,rbNotificationEventQueueDL 是一个死信队列,这个死信队列 会在消息超时 的时候,自动转发到指定的队列。定义了一个消费者来消费rbNotificationEventQueue 里面的消息。

 

 

纯配置类的配置方式:

@Configuration
public class RabbitMQConfig2 {

	public static final String sys_exchange  = "sys_exchange";

    
    public static final String queueMessageSysProfitRecordDay = "queueMessageSysProfitRecordDay";


    /**
     * 死信
     */
    public static final String queueMessageSysProfitRecordDay_DL = "queueMessageSysProfitRecordDay_DL";

    
    /**
     * 交换
     * @return
     */
    @Bean
    public DirectExchange sysExchange() {
        return new DirectExchange(RabbitMQConfig2.sys_exchange, true, false);
    }

    
    /**
     * 正常的队列
     * @return
     */
    @Bean
    public Queue queueMessageSysProfitRecordDay() {
    	return new Queue( RabbitMQConfig2.queueMessageSysProfitRecordDay );
    }

    /**
     *  死信队列
     */
    
    @Bean
    public Queue queueMessageSysProfitRecordDayDL() {
    	Map<String, Object> arguments = new HashMap<>();
		arguments.put("x-dead-letter-exchange", RabbitMQConfig2.sys_exchange);
		arguments.put("x-dead-letter-routing-key", RabbitMQConfig2.queueMessageSysProfitRecordDay);
		return new Queue(RabbitMQConfig2.queueMessageSysProfitRecordDay_DL, true, false, false, arguments);
    }

    
    /**
     * 绑定
     */
    @Bean
    public Binding  bindingMessageSysProfitRecordDay(Queue queueMessageSysProfitRecordDay ,DirectExchange sysExchange ) {
        return BindingBuilder.bind( queueMessageSysProfitRecordDay ).to( sysExchange ).with( RabbitMQConfig2.queueMessageSysProfitRecordDay );
    }

    
    /**
     * 死信绑定
     */ 
    
    @Bean
    public Binding  bindingMessageSysProfitRecordDayDL(Queue queueMessageSysProfitRecordDayDL ,DirectExchange sysExchange ) {
        return BindingBuilder.bind( queueMessageSysProfitRecordDayDL ).to( sysExchange ).with( RabbitMQConfig2.queueMessageSysProfitRecordDay_DL );
    }

}

 

 

附带发送消息和延时消息的代码:

@Component
public class MqMessageService{

	@Autowired
	private AmqpTemplate rabbitTemplate;
	
	/**
	 * 发送消息
	 * @param message
	 * @return
	 */
	public boolean send(  Message message ) {
		try {
			if( MessageType.QUEUE.equals( message.getMsgType() ) ) {
				rabbitTemplate.convertAndSend( message.getQueue() , JSONObject.toJSONString(message) );
				return true;
			}else if( MessageType.EXCHANGE.equals( message.getMsgType() ) ) {
				sendExchangeMessage( message );
				return true;
			}else {
				System.out.println( "暂时不处理" );
				return true;
			}
		}catch (Exception e) {
			System.out.println( e.getStackTrace() );
			return false;
		}
	}
	
	/**
	 *  发送Exchange的消息
	 * @param message
	 */
	private void sendExchangeMessage(  Message message  ) {
		if( message.getDelayTime() != null ) {
			String delayTime = message.getDelayTime().toString();
			MessagePostProcessor processor = new MessagePostProcessor(){
				@Override
				public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
					message.getMessageProperties().setExpiration( delayTime );
			        return message;
				}
		    };
			rabbitTemplate.convertAndSend( message.getExchange() , message.getRoutingKey() , JSONObject.toJSONString(message) ,processor );
			return;
		}
		rabbitTemplate.convertAndSend( message.getExchange() , message.getRoutingKey() , JSONObject.toJSONString(message) );
	}

}

  

附带消息对象:

public class Message {
	private Long id;
	
	private String exchange;
	
	private String routingKey;

	private String queue;
	
	private MessageType msgType;
	
	/**
	 * 只有 exchange 类型的有延时
	 */
	private Long  delayTime;

	public Message() {
		this.id = SnGeneratorUtil.getId();
	}
	
	public Long getId() {
		return id;
	}

	@Override
	public String toString() {
		return "Message [id=" + id + ", exchange=" + exchange + ", routingKey=" + routingKey + ", queue=" + queue
				+ ", msgType=" + msgType + ", delayTime=" + delayTime + "]";
	}

	public void setId(Long id) {
		this.id = id;
	}

	public String getExchange() {
		return exchange;
	}

	public void setExchange(String exchange) {
		this.exchange = exchange;
	}

	public String getRoutingKey() {
		return routingKey;
	}

	public void setRoutingKey(String routingKey) {
		this.routingKey = routingKey;
	}

	public String getQueue() {
		return queue;
	}

	public void setQueue(String queue) {
		this.queue = queue;
	}

	public MessageType getMsgType() {
		return msgType;
	}

	public void setMsgType(MessageType msgType) {
		this.msgType = msgType;
	}

	/**
	 * 只有 exchange 类型的有延时
	 */
	public Long getDelayTime() {
		return delayTime;
	}

	public void setDelayTime(Long delayTime) {
		this.delayTime = delayTime;
	}

}

  附带消息类型枚举:

public enum MessageType {

	
	QUEUE("队列"),
	EXCHANGE("交换机");

    private String mark;

    MessageType(String mark ) {
        this.mark = mark;
    }

    public String getMark() {
        return mark;
    }
}

  附带发送消息代码: 备注,这是发了一个上面的message 的子类消息 ,需要自定义。并且这个消息是延时10秒的。

MessageSysProfitRecordDay message = new MessageSysProfitRecordDay();
		message.setMsgType( MessageType.EXCHANGE );
		message.setExchange( RabbitMQConfig.sys_exchange );
		message.setRoutingKey( RabbitMQConfig.queueMessageSysProfitRecordDay_DL );
		
		message.setSourceId(sourceId);
		message.setYear(year);
		message.setMonth(month);
		message.setDay(day);
		message.setType( type.name() );
		message.setAmountType( amountType.name() );
		message.setAmount( amount );
		message.setDelayTime( 10000L );
		
        mqMessageService.send( message );

  附带接受消息的代码:

  

	@RabbitListener(queues = RabbitMQConfig.queueMessageSysProfitRecordDay)
	@RabbitHandler
	public void orderAward(String messageStr) {
		MessageSysProfitRecordDay message = null;
		if( logger.isInfoEnabled() ) {
			logger.info("收到一天系统日结算:{}",messageStr );
		}
	}

  

 

备注:spring boot集成的 自动 ack(在没抛出异常的情况下),别的方式需要手动ack。