摘要：
本篇博文是“Java秒杀系统实战系列文章”的第八篇，在这篇文章中我们将整合消息中间件RabbitMQ，包括添加依赖、加入配置信息以及自定义注入相关操作组件，比如RabbitTemplate等等，最终初步实现消息的发送和接收，并在下一篇章将其与邮件服务整合，实现“用户秒杀成功发送邮件通知消息”的功能！
内容：
对于消息中间件RabbitMQ，想必各位小伙伴没有用过、也该有听过，它是一款目前市面上应用相当广泛的消息中间件，可以实现消息异步通信、业务服务模块解耦、接口限流、消息分发等功能，在微服务、分布式系统架构中可以说是充当着一名了不起的角色！（详细的介绍，Debug在这里就不赘述了，各位小伙伴可以上官网看看其更多的介绍及其典型的应用场景）！
在本篇博文中，我们将使用RabbitMQ充当消息发送的组件，将它与后面篇章介绍的“邮件服务”结合实现“用户秒杀成功后异步发送邮件通知消息，告知用户秒杀已经成功！”，下面我们一起进入代码实战吧。
（1）要使用RabbitMQ，前提得在本地开发环境或者服务器安装RabbitMQ服务，如下图所示为Debug在本地安装RabbitMQ服务成功后访问其后端控制台应用的首页：
image
之后我们开始将其与SpringBoot进行整合。首先需要加入其依赖，其版本号跟SpringBoot的版本一致，版本号为1.5.7.RELEASE：
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
<version>${spring-boot.version}</version>
</dependency>
然后需要在配置文件application.properties中加入RabbitMQ服务相关的配置，比如其服务所在的Host、端口Port等等：
#rabbitmq
spring.rabbitmq.virtual-host=/
spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.listener.simple.concurrency=5
spring.rabbitmq.listener.simple.max-concurrency=15
spring.rabbitmq.listener.simple.prefetch=10
（2）紧接着，我们借助SpringBoot天然具有的一些特性，自动注入RabbitMQ一些组件的配置，包括其“单一实例消费者”配置、“多实例消费者”配置以及用于发送消息的操作组件实例“RabbitTemplate”的配置：
//通用化 Rabbitmq 配置
@Configuration
public class RabbitmqConfig {
private final static Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);
@Autowired
private Environment env;
@Autowired
private CachingConnectionFactory connectionFactory;
@Autowired
private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;
//单一消费者
@Bean(name = "singleListenerContainer")
public SimpleRabbitListenerContainerFactory listenerContainer(){
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
factory.setConcurrentConsumers(1);
factory.setMaxConcurrentConsumers(1);
factory.setPrefetchCount(1);
factory.setTxSize(1);
return factory;
}
//多个消费者
@Bean(name = "multiListenerContainer")
public SimpleRabbitListenerContainerFactory multiListenerContainer(){
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factoryConfigurer.configure(factory,connectionFactory);
factory.setMessageConverter(new Jackson2JsonMessageConverter());
//确认消费模式-NONE
factory.setAcknowledgeMode(AcknowledgeMode.NONE);
factory.setConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.concurrency",int.class));
factory.setMaxConcurrentConsumers(env.getProperty("spring.rabbitmq.listener.simple.max-concurrency",int.class));
factory.setPrefetchCount(env.getProperty("spring.rabbitmq.listener.simple.prefetch",int.class));
return factory;
}
@Bean
public RabbitTemplate rabbitTemplate(){
connectionFactory.setPublisherConfirms(true);
connectionFactory.setPublisherReturns(true);
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
}
});
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.warn("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
}
});
return rabbitTemplate;
}
}
在RabbitMQ的消息发送组件RabbitTemplate的配置中，我们还特意加入了“消息发送确认”、“消息丢失回调”的输出配置，即当消息正确进入到队列后，即代表消息发送成功；当消息找不到对应的队列（在某种程度上，其实也就是找不到交换机和路由）时，会输出消息丢失。
（3）完了之后，我们准备开始使用RabbitMQ实现消息的发送和接收。首先，我们需要在RabbitmqConfig配置类中创建队列、交换机、路由以及绑定等Bean组件，如下所示：
//构建异步发送邮箱通知的消息模型
@Bean
public Queue successEmailQueue(){
return new Queue(env.getProperty("mq.kill.item.success.email.queue"),true);
}
@Bean
public TopicExchange successEmailExchange(){
return new TopicExchange(env.getProperty("mq.kill.item.success.email.exchange"),true,false);
}
@Bean
public Binding successEmailBinding(){
return BindingBuilder.bind(successEmailQueue()).to(successEmailExchange()).with(env.getProperty("mq.kill.item.success.email.routing.key"));
}
其中，环境变量实例env读取的那些属性我们是配置在application.properties文件中的，如下所示：
mq.env=test
#秒杀成功异步发送邮件的消息模型
mq.kill.item.success.email.queue=${mq.env}.kill.item.success.email.queue
mq.kill.item.success.email.exchange=${mq.env}.kill.item.success.email.exchange
mq.kill.item.success.email.routing.key=${mq.env}.kill.item.success.email.routing.key
紧接着，我们需要在通用的消息发送服务类 RabbitSenderService 中写一段发送消息的方法，该方法用于接收“订单编号”参数，然后在数据库中查询其对应的详细订单记录，将该记录充当“消息”并发送至RabbitMQ的队列中，等待被监听消费：
/**
* RabbitMQ通用的消息发送服务
* @Author:debug (SteadyJack)
* @Date: 2019/6/21 21:47
**/
@Service
public class RabbitSenderService {
public static final Logger log= LoggerFactory.getLogger(RabbitSenderService.class);
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private Environment env;
@Autowired
private ItemKillSuccessMapper itemKillSuccessMapper;
//秒杀成功异步发送邮件通知消息
public void sendKillSuccessEmailMsg(String orderNo){
log.info("秒杀成功异步发送邮件通知消息-准备发送消息：{}",orderNo);
try {
if (StringUtils.isNotBlank(orderNo)){
KillSuccessUserInfo info=itemKillSuccessMapper.selectByCode(orderNo);
if (info!=null){
//TODO:rabbitmq发送消息的逻辑
rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
rabbitTemplate.setExchange(env.getProperty("mq.kill.item.success.email.exchange"));
rabbitTemplate.setRoutingKey(env.getProperty("mq.kill.item.success.email.routing.key"));
//TODO：将info充当消息发送至队列
rabbitTemplate.convertAndSend(info, new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
MessageProperties messageProperties=message.getMessageProperties();
messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
messageProperties.setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME,KillSuccessUserInfo.class);
return message;
}
});
}
}
}catch (Exception e){
log.error("秒杀成功异步发送邮件通知消息-发生异常，消息为：{}",orderNo,e.fillInStackTrace());
}
}
}
（4）最后，是在通用的消息接收服务类RabbitReceiverService中实现消息的接收，其完整的源代码如下所示：
/**
* RabbitMQ通用的消息接收服务
* @Author:debug (SteadyJack)
* @Date: 2019/6/21 21:47
**/
@Service
public class RabbitReceiverService {
public static final Logger log= LoggerFactory.getLogger(RabbitReceiverService.class);
@Autowired
private MailService mailService;
@Autowired
private Environment env;
@Autowired
private ItemKillSuccessMapper itemKillSuccessMapper;
//秒杀异步邮件通知-接收消息
@RabbitListener(queues = {"${mq.kill.item.success.email.queue}"},containerFactory = "singleListenerContainer")
public void consumeEmailMsg(KillSuccessUserInfo info){
try {
log.info("秒杀异步邮件通知-接收消息:{}",info);
//到时候这里将整合邮件服务发送邮件通知消息的逻辑
}catch (Exception e){
log.error("秒杀异步邮件通知-接收消息-发生异常：",e.fillInStackTrace());
}
}
}
至此，关于SpringBoot整合消息中间件RabbitMQ的代码实战，本篇文章就介绍到这里了。
最后一点，我们需要进行测试，即用户在界面发起“抢购”的请求操作之后，如果能秒杀成功，则RabbitMQ会发送、接收一条消息，如下所示：
image
好了，关于RabbitMQ的使用，本文到此就暂且告一段落了，在下一篇文章中我们将把它与邮件服务进行整合，实现“用户秒杀成功后异步发送邮件通知消息给到用户邮箱”的功能！除此之外，我们还将在后面的篇章介绍“如何使用RabbitMQ的死信队列，处理用户下单成功后却超时未支付的订单~在那里我们将采取失效的操作”。
补充：
由于相应的博客的更新可能并不会很快，故而如果有想要快速入门以及实战整套系统的，可以阅读： Java商城秒杀系统的设计与实战视频教程（SpringBoot版）
网友评论