目錄
0.交換機種類和區(qū)別
1.聲明隊列和交換機以及RountKey
2.初始化循環(huán)綁定
3.聲明交換機
4.監(jiān)聽隊列
4.1 監(jiān)聽普通隊列
4.2監(jiān)聽死信隊列
?5.削峰填谷的實現(xiàn)
0.交換機種類和區(qū)別
-
Direct Exchange（直連交換機）：
- 直連交換機將消息發(fā)送到與消息的路由鍵完全匹配的隊列。它是最簡單的交換機類型之一。
- 當(dāng)一個隊列使用某個直連交換機綁定時，它需要指定一個綁定鍵（binding key），當(dāng)消息的路由鍵與該綁定鍵完全匹配時，消息會被發(fā)送到該隊列。
-
Fanout Exchange（扇出交換機）：
- 扇出交換機會將消息發(fā)送到與其綁定的所有隊列，忽略消息的路由鍵。
- 當(dāng)一個隊列使用扇出交換機綁定時，它會接收到交換機發(fā)送的所有消息，無論消息的路由鍵是什么。
-
Topic Exchange（主題交換機）：
- 主題交換機根據(jù)消息的路由鍵和綁定鍵之間的模式匹配來路由消息。
- 綁定鍵可以使用通配符進行匹配，支持 '*' 匹配一個單詞，'#' 匹配零個或多個單詞，從而允許更靈活的路由規(guī)則。
-
Headers Exchange（標(biāo)頭交換機）：
- 標(biāo)頭交換機根據(jù)消息的標(biāo)頭（headers）中的鍵值對來路由消息，而不是使用路由鍵。
- 在將隊列綁定到標(biāo)頭交換機時，可以指定一組標(biāo)頭鍵值對，只有當(dāng)消息的標(biāo)頭中包含與綁定相匹配的所有鍵值對時，消息才會被路由到該隊列。
如果滿足key的前提下，綁定同一個交換機的隊列都會分配到相同數(shù)量的信息
比如此時交換機有20條信息，a，b隊列都會分配到20條信息
默認(rèn)情況下，會輪詢分配給消費者，也可以設(shè)置最多獲取多少條未被消費的信息，根據(jù)消費者的消費能力來設(shè)置
1.聲明隊列和交換機以及RountKey
package com.example.config; import lombok.Getter; @Getter public enum RabbitmqBind { DATA_CLEAN_PROCESS_DEAD( RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP, RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS_DEAD, RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS_DEAD, false, false, null, null ), DATA_CLEAN_PROCESS( RabbitMqExchangeEnum.E_DIRECT_RCP, RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS, RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS, true, true, RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP, RabbitmqRoutingKey.K_DATA_CLEAN_PROCESS_DEAD), SMS_CLEAN_DEAD( RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP, RabbitMqQueueConstants.Q_API_TO_DCN_SMS_DEAD, RabbitmqRoutingKey.K_DATA_CLEAN_FINISH_DEAD, true, false, null, null ), SMS_CLEAN( RabbitMqExchangeEnum.E_TOPIC_RCP, RabbitMqQueueConstants.Q_API_TO_DCN_SMS, RabbitmqRoutingKey.K_API_TO_DCN_SMS, true, true, RabbitMqExchangeEnum.DEAD_E_DIRECT_RCP, RabbitmqRoutingKey.K_DATA_CLEAN_FINISH_DEAD ), ; /** * 交換機 */ private RabbitMqExchangeEnum exchange; /** * 隊列名稱 */ private String queueName; /** * 路由Key */ private RabbitmqRoutingKey routingKey; /** * 綁定標(biāo)識 * 是否啟用 */ private Boolean isBind; /** * 是否綁定死信 */ private Boolean isDeathBelief; /** * 綁定的死信交換機 */ private RabbitMqExchangeEnum boundDeadExchange; /** * 死信key */ private RabbitmqRoutingKey deadRoutingKey; RabbitmqBind(RabbitMqExchangeEnum exchange, String queueName, RabbitmqRoutingKey routingKey, Boolean isBind, Boolean isDeathBelief, RabbitMqExchangeEnum boundDeadExchange, RabbitmqRoutingKey deadRoutingKey ) { this.exchange = exchange; this.queueName = queueName; this.routingKey = routingKey; this.isBind = isBind; this.isDeathBelief = isDeathBelief; this.boundDeadExchange = boundDeadExchange; this.deadRoutingKey = deadRoutingKey; } /** * 交換機 */ @Getter public enum RabbitMqExchangeEnum { /** * 交換機定義，類型 - 名稱 */ E_DIRECT_RCP("direct", "E_DIRECT_RCP"), DEAD_E_DIRECT_RCP("direct", "DEAD_E_DIRECT_RCP"), E_TOPIC_RCP("topic", "E_TOPIC_RCP"), E_TOPIC_PAY("topic", "E_TOPIC_PAY"); private String exchangeType; private String exchangeName; RabbitMqExchangeEnum(String exchangeType, String exchangeName) { this.exchangeType = exchangeType; this.exchangeName = exchangeName; } } /** * 隊列名定義 */ public interface RabbitMqQueueConstants { /** * 接收清洗數(shù)據(jù) */ String Q_DATA_CLEAN_PROCESS = "RMPS_TO_RCP_DATA_CLEAN_PROCESS"; /** * 清洗結(jié)束通知 */ String Q_API_TO_DCN_SMS = "Q_API_TO_DCN_SMS"; /** * 死信隊列 */ String Q_DATA_CLEAN_PROCESS_DEAD = "Q_DATA_CLEAN_PROCESS_DEAD"; /** * 清洗結(jié)束通知死信隊列 */ String Q_API_TO_DCN_SMS_DEAD = "Q_API_TO_DCN_SMS_DEAD"; } /** * routingKey */ @Getter public enum RabbitmqRoutingKey { /** * 路由 */ K_DATA_CLEAN_PROCESS("K_DATA_CLEAN_PROCESS"), K_API_TO_DCN_SMS("K_API_TO_DCN_SMS"), // 路由綁定死信路由 DEAD("DEAD"), //死信路由 K_DATA_CLEAN_PROCESS_DEAD("K_DATA_CLEAN_PROCESS_DEAD"), K_DATA_CLEAN_FINISH_DEAD("K_DATA_CLEAN_FINISH_DEAD"), ; private String keyName; RabbitmqRoutingKey(String keyName) { this.keyName = keyName; } } }
2.初始化循環(huán)綁定
package com.example.config; import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.Exchange; import org.springframework.amqp.core.Queue; import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.rabbit.annotation.EnableRabbit; import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.transaction.RabbitTransactionManager; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Lazy; import javax.annotation.PostConstruct; import javax.annotation.Resource; import java.util.Arrays; @Configuration @ConditionalOnClass(EnableRabbit.class) public class MqConfig { @Resource protected RabbitTemplate rabbitTemplate; @Resource ConnectionFactory connectionFactory; // // @Lazy // @Autowired // protected RabbitAdmin rabbitAdmin; // // // public static final int DEFAULT_CONCURRENT = 10; // // @Bean("customContainerFactory") // public SimpleRabbitListenerContainerFactory containerFactory(SimpleRabbitListenerContainerFactoryConfigurer configurer, // ConnectionFactory connectionFactory) { // SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); // factory.setConcurrentConsumers(DEFAULT_CONCURRENT); // factory.setMaxConcurrentConsumers(DEFAULT_CONCURRENT); // configurer.configure(factory, connectionFactory); // return factory; // } // // @Bean // @ConditionalOnMissingBean // public RabbitTransactionManager rabbitTransactionManager(ConnectionFactory connectionFactory) { // return new RabbitTransactionManager(connectionFactory); // } // // @Bean // @ConditionalOnMissingBean // public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) { // return new RabbitAdmin(connectionFactory); // } @PostConstruct protected void init() { RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory); rabbitTemplate.setChannelTransacted(true); //創(chuàng)建exchange Arrays.stream(RabbitmqBind.RabbitMqExchangeEnum.values()) .forEach(rabbitMqExchangeEnum -> { Exchange exchange = RabbitmqExchange .getInstanceByType(rabbitMqExchangeEnum.getExchangeType()) .createExchange(rabbitMqExchangeEnum.getExchangeName()); rabbitAdmin.declareExchange(exchange); } ); //創(chuàng)建隊列并綁定exchange Arrays.stream(RabbitmqBind.values()).forEach(RabbitmqBind -> { if (RabbitmqBind.getIsBind()) { if (RabbitmqBind.getIsDeathBelief()) { //需要綁定死信交換機的隊列 rabbitAdmin.declareQueue(QueueBuilder.durable(RabbitmqBind.getQueueName()) .ttl(60000).deadLetterExchange(RabbitmqBind.getBoundDeadExchange().getExchangeName()) .deadLetterRoutingKey(RabbitmqBind.getDeadRoutingKey().getKeyName()).build()); rabbitAdmin.declareBinding(new Binding(RabbitmqBind.getQueueName(), Binding.DestinationType.QUEUE, RabbitmqBind.getExchange().getExchangeName(), RabbitmqBind.getRoutingKey().getKeyName(), null)); } else { //不需要綁定死信交換機的隊列 rabbitAdmin.declareQueue(new Queue(RabbitmqBind.getQueueName(), true, false, false, null)); rabbitAdmin.declareBinding(new Binding(RabbitmqBind.getQueueName(), Binding.DestinationType.QUEUE, RabbitmqBind.getExchange().getExchangeName(), RabbitmqBind.getRoutingKey().getKeyName(), null)); } } }); } }?綁定的形式由枚舉類中定義
3.聲明交換機
package com.example.config;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.TopicExchange;
import java.util.Arrays;
@Getter
@Slf4j
public enum RabbitmqExchange {
DIRECT("direct"){
@Override
public Exchange createExchange(String exchangeName) {
return new DirectExchange(exchangeName, true, false);
}
},
TOPIC("topic"){
@Override
public Exchange createExchange(String exchangeName) {
return new TopicExchange(exchangeName, true, false);
}
};
public static RabbitmqExchange getInstanceByType(String type){
return Arrays.stream(RabbitmqExchange.values()).filter(e -> e.getType().equals(type))
.findAny()
.orElseThrow(() ->
// new ProcessException("無效的exchange type")
new RuntimeException("無效的exchange type")
);
}
private String type;
RabbitmqExchange(String type) {
this.type = type;
}
public abstract Exchange createExchange(String exchangeName);
}
4.監(jiān)聽隊列
4.1 監(jiān)聽普通隊列
package com.example.listener; import com.example.config.RabbitmqBind; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang.StringUtils; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Slf4j @Component @RabbitListener(queues = { RabbitmqBind.RabbitMqQueueConstants.Q_DATA_CLEAN_PROCESS }, concurrency = "1-5") //, containerFactory = "customContainerFactory" public class MqListener { @RabbitHandler public void processMessage(String message) { log.info("DataClean recive message :{} ", message); process(message); } @RabbitHandler public void processMessage(byte[] message) { String msg = new String(message); log.info("DataClean recive message :{} ", msg); process(msg); } /** * 處理推送消息 * @param message */ private void process(String message) { log.info("process message :{}" , message); if(StringUtils.isBlank(message)) { log.error("process message is blank , message:{}" , message); return; } } }?監(jiān)聽并處理任務(wù)
4.2監(jiān)聽死信隊列
package com.example.listener; import com.example.config.RabbitmqBind; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang.StringUtils; import org.springframework.amqp.rabbit.annotation.RabbitHandler; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; @Slf4j @Component @RabbitListener(queues = { RabbitmqBind.RabbitMqQueueConstants.Q_API_TO_DCN_SMS_DEAD }, concurrency = "1-5") public class DeadListener { @RabbitHandler public void processMessage(String message) { log.info("DataClean recive message :{} ", message); process(message); } @RabbitHandler public void processMessage(byte[] message) { String msg = new String(message); log.info("DataClean recive message :{} ", msg); process(msg); } /** * 處理推送消息 * @param message */ private void process(String message) { log.info("Dead process message :{}" , message); if(StringUtils.isBlank(message)) { log.error("Dead process message is blank , message:{}" , message); return; } } }
?5.削峰填谷的實現(xiàn)
把高峰期的消息填進低峰期
可以用拉取的方式來實現(xiàn)文章來源：http://www.zghlxwxcb.cn/news/detail-856444.html
或者用消費者的最大數(shù)量和最小數(shù)量來實現(xiàn)文章來源地址http://www.zghlxwxcb.cn/news/detail-856444.html
channel.basicQos();//設(shè)置最大獲取未確認(rèn)消息的數(shù)量，實現(xiàn)權(quán)重
到了這里，關(guān)于消息隊列中間件，RabbitMQ的使用，死信隊列，延遲隊列，利用枚舉實現(xiàn)隊列，交換機，RountKey的聲明的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
