Spring Boot整合RabbitMQ詳細(xì)教程

org.springframework.boot
spring-boot-starter-amqp
1.5.2.RELEASE

#對于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)者到消費者的路線，保證數(shù)據(jù)能按照指定的方式進(jìn)行傳輸,
Exchange：消息交換機(jī),它指定消息按什么規(guī)則,路由到哪個隊列。
Queue:消息的載體,每個消息都會被投到一個或多個隊列。
Binding:綁定，它的作用就是把exchange和queue按照路由規(guī)則綁定起來.
Routing Key:路由關(guān)鍵字,exchange根據(jù)這個關(guān)鍵字進(jìn)行消息投遞。
vhost:虛擬主機(jī),一個broker里可以有多個vhost，用作不同用戶的權(quán)限分離。
Producer:消息生產(chǎn)者,就是投遞消息的程序.
Consumer:消息消費者,就是接受消息的程序.
Channel:消息通道,在客戶端的每個連接里,可建立多個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，所以不能自動注入
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對應(yīng)的隊列當(dāng)中去，對應(yīng)的是隊列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("消息成功消費");
}?else?{
logger.info("消息消費失敗:"?+?cause);
}
}
}

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

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

@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("接收處理隊列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接收處理隊列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接收處理隊列A當(dāng)中的消息：?"?+?content);
}
?
}

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

//配置fanout_exchange
@Bean
FanoutExchange?fanoutExchange()?{
return?new?FanoutExchange(RabbitConfig.FANOUT_EXCHANGE);
}
?
//把所有的隊列都綁定到這個交換機(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);
}