rabbit路由源码分析

三种路由模式

rabbitMq有三种交换机路由模式 https://my.oschina.net/huangcongmin12/blog/885345

 
任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上
1.这种模式较为复杂&＃xff0c;简单来说&＃xff0c;就是每个队列都有其关心的主题&＃xff0c;所有的消息都带有一个“标题”(RouteKey)&＃xff0c;Exchange会将消息转发到所有关注主题能与RouteKey模糊匹配的队列。
2.这种模式需要RouteKey&＃xff0c;也许要提前绑定Exchange与Queue。
3.在进行绑定时&＃xff0c;要提供一个该队列关心的主题&＃xff0c;如“#.log.#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。
4.“#”表示0个或若干个关键字&＃xff0c;“*”表示一个关键字。如“log.*”能与“log.warn”匹配&＃xff0c;无法与“log.warn.timeout”匹配&＃xff1b;但是“log.#”能与上述两者匹配。
5.同样&＃xff0c;如果Exchange没有发现能够与RouteKey匹配的Queue&＃xff0c;则会抛弃此消息。

数据结构

每一个队列有一个consumer来维护

consumer的数据结构是

queue: 消息

queues:队列

源码

接受消息 

private boolean doReceiveAndExecute(BlockingQueueConsumer consumer) throws Throwable { //NOSONARChannel channel &＃61; consumer.getChannel();for (int i &＃61; 0; i //取出消息出来&＃xff0c;consumer是一个封装&＃xff0c;对所有连接信息&＃xff0c;channel的封装logger.trace("Waiting for message from consumer.");Message message &＃61; consumer.nextMessage(this.receiveTimeout);if (message &＃61;&＃61; null) {break;}try {executeListener(channel, message);}catch (ImmediateAcknowledgeAmqpException e) {if (this.logger.isDebugEnabled()) {this.logger.debug("User requested ack for failed delivery: "&＃43; message.getMessageProperties().getDeliveryTag());}break;}catch (Throwable ex) { //NOSONARif (causeChainHasImmediateAcknowledgeAmqpException(ex)) {if (this.logger.isDebugEnabled()) {this.logger.debug("User requested ack for failed delivery: "&＃43; message.getMessageProperties().getDeliveryTag());}break;}if (this.transactionManager !&＃61; null) {if (this.transactionAttribute.rollbackOn(ex)) {RabbitResourceHolder resourceHolder &＃61; (RabbitResourceHolder) TransactionSynchronizationManager.getResource(getConnectionFactory());if (resourceHolder !&＃61; null) {consumer.clearDeliveryTags();}else {/** If we don&＃39;t actually have a transaction, we have to roll back* manually. See prepareHolderForRollback().*/consumer.rollbackOnExceptionIfNecessary(ex);}throw ex; // encompassing transaction will handle the rollback.}else {if (this.logger.isDebugEnabled()) {this.logger.debug("No rollback for " &＃43; ex);}break;}}else {consumer.rollbackOnExceptionIfNecessary(ex);throw ex;}}}return consumer.commitIfNecessary(isChannelLocallyTransacted(channel));}

consumer处理消息投递

&＃64;Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)throws IOException {if (logger.isDebugEnabled()) {logger.debug("Storing delivery for consumerTag: &＃39;"&＃43; consumerTag &＃43; "&＃39; with deliveryTag: &＃39;" &＃43; envelope.getDeliveryTag() &＃43; "&＃39; in "&＃43; BlockingQueueConsumer.this);}try {if (BlockingQueueConsumer.this.abortStarted > 0) {if (!BlockingQueueConsumer.this.queue.offer(new Delivery(consumerTag, envelope, properties, body),BlockingQueueConsumer.this.shutdownTimeout, TimeUnit.MILLISECONDS)) {RabbitUtils.setPhysicalCloseRequired(getChannel(), true);// Defensive - should never happenBlockingQueueConsumer.this.queue.clear();getChannel().basicNack(envelope.getDeliveryTag(), true, true);getChannel().basicCancel(consumerTag);try {getChannel().close();}catch (TimeoutException e) {// no-op}}}else {BlockingQueueConsumer.this.queue.put(new Delivery(consumerTag, envelope, properties, body));}}catch (InterruptedException e) {Thread.currentThread().interrupt();}}