Java秒杀系统实战系列~整合RabbitMQ实现消息异步发送

简介: 在这篇文章中我们将整合消息中间件RabbitMQ,包括添加依赖、加入配置信息以及自定义注入相关操作组件,比如RabbitTemplate等等,最终初步实现消息的发送和接收,并在下一篇章将其与邮件服务整合,实现“用户秒杀成功发送邮件通知消息”的功能!

摘要:

本篇博文是“Java秒杀系统实战系列文章”的第八篇,在这篇文章中我们将整合消息中间件RabbitMQ,包括添加依赖、加入配置信息以及自定义注入相关操作组件,比如RabbitTemplate等等,最终初步实现消息的发送和接收,并在下一篇章将其与邮件服务整合,实现“用户秒杀成功发送邮件通知消息”的功能!

内容:

对于消息中间件RabbitMQ,想必各位小伙伴没有用过、也该有听过,它是一款目前市面上应用相当广泛的消息中间件,可以实现消息异步通信、业务服务模块解耦、接口限流、消息分发等功能,在微服务、分布式系统架构中可以说是充当着一名了不起的角色!(详细的介绍,Debug在这里就不赘述了,各位小伙伴可以上官网看看其更多的介绍及其典型的应用场景)!

在本篇博文中,我们将使用RabbitMQ充当消息发送的组件,将它与后面篇章介绍的“邮件服务”结合实现“用户秒杀成功后异步发送邮件通知消息,告知用户秒杀已经成功!”,下面我们一起进入代码实战吧。

(1)要使用RabbitMQ,前提得在本地开发环境或者服务器安装RabbitMQ服务,如下图所示为Debug在本地安装RabbitMQ服务成功后访问其后端控制台应用的首页:
image
之后我们开始将其与SpringBoot进行整合。首先需要加入其依赖,其版本号跟SpringBoot的版本一致,版本号为1.5.7.RELEASE:

<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的死信队列,处理用户下单成功后却超时未支付的订单~在那里我们将采取失效的操作”。

补充:

1、由于相应的博客的更新可能并不会很快,故而如果有想要快速入门以及实战整套系统的,可以阅读: Java商城秒杀系统的设计与实战视频教程(SpringBoot版)

2、目前,这一秒杀系统的整体构建与代码实战已经全部完成了,完整的源代码数据库地址可以来这里下载:https://gitee.com/steadyjack/SpringBoot-SecondKill

相关实践学习
消息队列RocketMQ版:基础消息收发功能体验
本实验场景介绍消息队列RocketMQ版的基础消息收发功能,涵盖实例创建、Topic、Group资源创建以及消息收发体验等基础功能模块。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
2月前
|
存储 Java 开发者
Java Map实战:用HashMap和TreeMap轻松解决复杂数据结构问题!
【10月更文挑战第17天】本文深入探讨了Java中HashMap和TreeMap两种Map类型的特性和应用场景。HashMap基于哈希表实现,支持高效的数据操作且允许键值为null;TreeMap基于红黑树实现,支持自然排序或自定义排序,确保元素有序。文章通过具体示例展示了两者的实战应用,帮助开发者根据实际需求选择合适的数据结构,提高开发效率。
64 2
|
2月前
|
消息中间件 监控 Java
RocketMQ 同步发送、异步发送和单向发送,如何选择?
本文详细分析了 RocketMQ 中同步发送、异步发送和单向发送三种消息发送方式的原理、优缺点及适用场景。同步发送可靠性高但延迟较大,适合订单系统等场景;异步发送非阻塞且延迟低,适用于实时数据处理等场景;单向发送高效但可靠性低,适用于日志收集等场景。文章还提供了示例代码和核心源码分析,帮助读者更好地理解每种发送方式的特点。
261 4
|
19天前
|
安全 Java 开发者
Java 多线程并发控制:深入理解与实战应用
《Java多线程并发控制:深入理解与实战应用》一书详细解析了Java多线程编程的核心概念、并发控制技术及其实战技巧,适合Java开发者深入学习和实践参考。
40 6
|
2月前
|
存储 消息中间件 安全
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
【10月更文挑战第9天】本文介绍了如何利用JUC组件实现Java服务与硬件通过MQTT的同步通信(RRPC)。通过模拟MQTT通信流程,使用`LinkedBlockingQueue`作为消息队列,详细讲解了消息发送、接收及响应的同步处理机制,包括任务超时处理和内存泄漏的预防措施。文中还提供了具体的类设计和方法实现,帮助理解同步通信的内部工作原理。
JUC组件实战:实现RRPC(Java与硬件通过MQTT的同步通信)
|
2月前
|
开发框架 Java 程序员
揭开Java反射的神秘面纱:从原理到实战应用!
本文介绍了Java反射的基本概念、原理及应用场景。反射允许程序在运行时动态获取类的信息并操作其属性和方法,广泛应用于开发框架、动态代理和自定义注解等领域。通过反射,可以实现更灵活的代码设计,但也需注意其性能开销。
52 1
|
3月前
|
缓存 负载均衡 Dubbo
Dubbo技术深度解析及其在Java中的实战应用
Dubbo是一款由阿里巴巴开源的高性能、轻量级的Java分布式服务框架,它致力于提供高性能和透明化的RPC远程服务调用方案,以及SOA服务治理方案。
84 6
|
3月前
|
Java
领略Lock接口的风采,通过实战演练,让你迅速掌握这门高深武艺,成为Java多线程领域的武林盟主
领略Lock接口的风采,通过实战演练,让你迅速掌握这门高深武艺,成为Java多线程领域的武林盟主
38 7
|
3月前
|
消息中间件 Kafka 数据安全/隐私保护
RabbitMQ异步通信详解
RabbitMQ异步通信详解
102 16
|
3月前
|
Java Android开发 C++
🚀Android NDK开发实战!Java与C++混合编程,打造极致性能体验!📊
在Android应用开发中,追求卓越性能是不变的主题。本文介绍如何利用Android NDK(Native Development Kit)结合Java与C++进行混合编程,提升应用性能。从环境搭建到JNI接口设计,再到实战示例,全面展示NDK的优势与应用技巧,助你打造高性能应用。通过具体案例,如计算斐波那契数列,详细讲解Java与C++的协作流程,帮助开发者掌握NDK开发精髓,实现高效计算与硬件交互。
158 1
|
2月前
|
消息中间件 数据采集 中间件
RabbitMQ的使用—实战
RabbitMQ的使用—实战
89 0