<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>1.5.2.RELEASE</version>
</dependency>

#對(duì)于rabbitMQ的支持
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

/**
Broker:它提供一種傳輸服務(wù),它的角色就是維護(hù)一條從生產(chǎn)者到消費(fèi)者的路線，保證數(shù)據(jù)能按照指定的方式進(jìn)行傳輸,
Exchange：消息交換機(jī),它指定消息按什么規(guī)則,路由到哪個(gè)隊(duì)列。
Queue:消息的載體,每個(gè)消息都會(huì)被投到一個(gè)或多個(gè)隊(duì)列。
Binding:綁定，它的作用就是把exchange和queue按照路由規(guī)則綁定起來(lái).
Routing Key:路由關(guān)鍵字,exchange根據(jù)這個(gè)關(guān)鍵字進(jìn)行消息投遞。
vhost:虛擬主機(jī),一個(gè)broker里可以有多個(gè)vhost，用作不同用戶的權(quán)限分離。
Producer:消息生產(chǎn)者,就是投遞消息的程序.
Consumer:消息消費(fèi)者,就是接受消息的程序.
Channel:消息通道,在客戶端的每個(gè)連接里,可建立多個(gè)channel.
*/
@Configuration
public class RabbitConfig {
 
private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
@Value("${spring.rabbitmq.host}")
private String host;
 
@Value("${spring.rabbitmq.port}")
private int port;
 
@Value("${spring.rabbitmq.username}")
private String username;
 
@Value("${spring.rabbitmq.password}")
private String password;
 
 
public static final String EXCHANGE_A = "my-mq-exchange_A";
public static final String EXCHANGE_B = "my-mq-exchange_B";
public static final String EXCHANGE_C = "my-mq-exchange_C";
 
 
public static final String QUEUE_A = "QUEUE_A";
public static final String QUEUE_B = "QUEUE_B";
public static final String QUEUE_C = "QUEUE_C";
 
public static final String ROUTINGKEY_A = "spring-boot-routingKey_A";
public static final String ROUTINGKEY_B = "spring-boot-routingKey_B";
public static final String ROUTINGKEY_C = "spring-boot-routingKey_C";
 
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host,port);
connectionFactory.setUsername(username);
connectionFactory.setPassword(password);
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true);
return connectionFactory;
}
 
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
//必須是prototype類型
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory());
return template;
}
}

@Component
public class MsgProducer implements RabbitTemplate.ConfirmCallback {
 
private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
//由于rabbitTemplate的scope屬性設(shè)置為ConfigurableBeanFactory.SCOPE_PROTOTYPE，所以不能自動(dòng)注入
private RabbitTemplate rabbitTemplate;
/**
* 構(gòu)造方法注入rabbitTemplate
*/
@Autowired
public MsgProducer(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
rabbitTemplate.setConfirmCallback(this); //rabbitTemplate如果為單例的話，那回調(diào)就是最后設(shè)置的內(nèi)容
}
 
public void sendMsg(String content) {
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
//把消息放入ROUTINGKEY_A對(duì)應(yīng)的隊(duì)列當(dāng)中去，對(duì)應(yīng)的是隊(duì)列A
rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE_A, RabbitConfig.ROUTINGKEY_A, content, correlationId);
}
/**
* 回調(diào)
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
logger.info(" 回調(diào)id:" + correlationData);
if (ack) {
logger.info("消息成功消費(fèi)");
} else {
logger.info("消息消費(fèi)失敗:" + cause);
}
}
}

/**
* 針對(duì)消費(fèi)者配置
* 1. 設(shè)置交換機(jī)類型
* 2. 將隊(duì)列綁定到交換機(jī)
FanoutExchange: 將消息分發(fā)到所有的綁定隊(duì)列，無(wú)routingkey的概念
HeadersExchange ：通過(guò)添加屬性key-value匹配
DirectExchange:按照routingkey分發(fā)到指定隊(duì)列
TopicExchange:多關(guān)鍵字匹配
*/
@Bean
public DirectExchange defaultExchange() {
return new DirectExchange(EXCHANGE_A);
}

/**
* 獲取隊(duì)列A
* @return
*/
@Bean
public Queue queueA() {
return new Queue(QUEUE_A, true); //隊(duì)列持久
}

@Bean
public Binding binding() {
 
return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);
}

@Bean
public Binding binding() {
return BindingBuilder.bind(queueA()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_A);
}
@Bean
public Binding bindingB(){
return BindingBuilder.bind(queueB()).to(defaultExchange()).with(RabbitConfig.ROUTINGKEY_B);
}

@Component
@RabbitListener(queues = RabbitConfig.QUEUE_A)
public class MsgReceiver {
 
private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
@RabbitHandler
public void process(String content) {
logger.info("接收處理隊(duì)列A當(dāng)中的消息： " + content);
}
 
}

@Component
@RabbitListener(queues = RabbitConfig.QUEUE_A)
public class MsgReceiverC_one {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
@RabbitHandler
public void process(String content) {
logger.info("處理器one接收處理隊(duì)列A當(dāng)中的消息： " + content);
}
}

@Component
@RabbitListener(queues = RabbitConfig.QUEUE_A)
public class MsgReceiverC_two {
private final Logger logger = LoggerFactory.getLogger(this.getClass());
@RabbitHandler
public void process(String content) {
logger.info("處理器two接收處理隊(duì)列A當(dāng)中的消息： " + content);
}
 
}

@Bean
public SimpleMessageListenerContainer messageContainer() {
//加載處理消息A的隊(duì)列
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
//設(shè)置接收多個(gè)隊(duì)列里面的消息，這里設(shè)置接收隊(duì)列A
//假如想一個(gè)消費(fèi)者處理多個(gè)隊(duì)列里面的信息可以如下設(shè)置：
//container.setQueues(queueA(),queueB(),queueC());
container.setQueues(queueA());
container.setExposeListenerChannel(true);
//設(shè)置最大的并發(fā)的消費(fèi)者數(shù)量
container.setMaxConcurrentConsumers(10);
//最小的并發(fā)消費(fèi)者的數(shù)量
container.setConcurrentConsumers(1);
//設(shè)置確認(rèn)模式手工確認(rèn)
container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
container.setMessageListener(new ChannelAwareMessageListener() {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
/**通過(guò)basic.qos方法設(shè)置prefetch_count=1，這樣RabbitMQ就會(huì)使得每個(gè)Consumer在同一個(gè)時(shí)間點(diǎn)最多處理一個(gè)Message，
換句話說(shuō),在接收到該Consumer的ack前,它不會(huì)將新的Message分發(fā)給它 */
channel.basicQos(1);
byte[] body = message.getBody();
logger.info("接收處理隊(duì)列A當(dāng)中的消息:" + new String(body));
/**為了保證永遠(yuǎn)不會(huì)丟失消息，RabbitMQ支持消息應(yīng)答機(jī)制。
當(dāng)消費(fèi)者接收到消息并完成任務(wù)后會(huì)往RabbitMQ服務(wù)器發(fā)送一條確認(rèn)的命令，然后RabbitMQ才會(huì)將消息刪除。*/
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
});
return container;
}

//配置fanout_exchange
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange(RabbitConfig.FANOUT_EXCHANGE);
}
 
//把所有的隊(duì)列都綁定到這個(gè)交換機(jī)上去
@Bean
Binding bindingExchangeA(Queue queueA,FanoutExchange fanoutExchange) {
return BindingBuilder.bind(queueA).to(fanoutExchange);
}
@Bean
Binding bindingExchangeB(Queue queueB, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(queueB).to(fanoutExchange);
}
@Bean
Binding bindingExchangeC(Queue queueC, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(queueC).to(fanoutExchange);
}

public void sendAll(String content) {
rabbitTemplate.convertAndSend("fanoutExchange","", content);
}

如有文章對(duì)你有幫助，
“在看”和轉(zhuǎn)發(fā)是對(duì)我最大的支持！

推薦， GitHub 書籍倉(cāng)庫(kù)
https://github.com/ebooklist/awesome-ebooks-list
整理了大部分常用 技術(shù)書籍PDF，持續(xù)更新中... 你需要的技術(shù)書籍，這里可能都有...

點(diǎn)擊文末“閱讀原文”可直達(dá)
整理不易，麻煩各位小伙伴在GitHub中來(lái)個(gè)Star支持一下