SpringBoot实战项目整合RabbitMQ+ElaticSearch实现SKU上下架功能

😊 @ 作者： Eric
💖 @ 主页： https://blog.csdn.net/weixin_47316183?type=blog
🎉 @ 主题：SpringBoot实战项目整合RabbitMQ+ElaticSearch实现SKU上下架功能
⏱️ @ 创作时间： 2023年07月03日

文章目录

前言

最终实现效果：针对SKU的上下架

上架效果：

1、后台选择

SKU

，点击上架，该

SKU

修改为

上架

状态
2、同时向

MQ

发送消息
3、服务监听收到消息后向

Es

中新增该

SKU

基本信息

下架效果：

1、后台选择

SKU

，点击下架，该

SKU

修改为

下架

状态
2、同时向

MQ

发送消息
3、服务监听收到消息后向

Es

中删除该

SKU

基本信息

那为什么一个上架的功能要使用这么多知识点呢，最简单的方式不就是修改SKU的状态，然后用户端只显示已上架的SKU就可以么。

原因有两点：

1.提高效率
2.还是效率

我们知道，在用户端搜索SKU信息，最简单的方式就是通过sku关键词直接模糊查询数据库中的sku信息，但是每次查询数据库都会发生IO的碰撞，显然不是最优的解决方法，而ElasticSearch就是为了检索而生，所以我们上架后自动将SKU数据存放到ElasticSearch中，然后用户端直接检索es中的数据即可，这样检索速度就哐哐哐的提上去了。

那为什么还要用到消息队列MQ呢，这是因为RabbitMQ是异步发送，当我们一个人操作sku可能没什么，那如果我们的系统有100万个商户，并且这100万个商户同时操作上下架呢？那么想想就是很恐怖的，当然，真到了这个量级肯定还需要再次调优。这个时候异步的重要的不言而喻。

主要使用技术点：

• JDK（版本为：1.8）
• SpringBoot（版本为：2.3.6.RELEASE）
• SpringData（用来操作ES）
• RabbitMQ（版本为：3.8）
• ElaticSearch（版本为：7.8）
• Spring Cloud（版本为：Hoxton.SR8）
• Nacos（版本为：2.2.3）

1、前置条件

1. Nacos的运行
2. ElasticSearch的运行（我这里使用的是IK分词器和Kibana工具）
3. RabbitMQ的运行
4. 项目使用的是SpringBoot

关于如何安装和运行这些软件大家可以去百度搜索下，很简单的。

2、搭建service-search模块

1、创建一个模块专门用来操作ES，根据自己项目接口来放，我这里放在service业务模块下

2、引入ES依赖

<dependencies><!-- ES依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId></dependency><!-- 引入远程调用模块 - 商品模块 --><dependency><groupId>com.atguigu</groupId><artifactId>service-product-client</artifactId><version>1.0-SNAPSHOT</version></dependency></dependencies>

3、es模块的配置文件

server:
  port: 8204

feign:
  sentinel:
    enabled: true
  client:
    config:
      default:   #配置全局的feign的调用超时时间  如果 有指定的服务配置 默认的配置不会生效
        connectTimeout: 30000# 指定的是 消费者 连接服务提供者的连接超时时间 是否能连接  单位是毫秒
        readTimeout: 50000# 指定的是调用服务提供者的 服务 的超时时间（）  单位是毫秒
spring:
  main:
    allow-bean-definition-overriding: true#当遇到同样名字的时候，是否允许覆盖注册
  elasticsearch:   # ElaticSearch
    rest:
      uris: http://localhost:9200
  rabbitmq:
    host: 192.168.64.109
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: CORRELATED
    publisher-returns: true
    listener:
      simple:
        prefetch: 1
        concurrency: 3
        acknowledge-mode: manual
  redis:
    host: localhost
    port: 6379
    database: 0
    timeout: 1800000
    password:
    lettuce:
      pool:
        max-active: 20#最大连接数
        max-wait: -1#最大阻塞等待时间(负数表示没限制)
        max-idle: 5#最大空闲
        min-idle: 0#最小空闲

4、记得在ES启动类上加上服务注册和远程调用注解
这里需要引入两个依赖，一个是

nacos服务注册

，一个是

服务调用openfeign

，因为这两个依赖我是放在了es模块的父工程中，所以es模块我是不需要引入的，大家根据自己项目结构来即可

@EnableDiscoveryClient      //服务注册
@EnableFeignClients         //服务调用

3、开发功能接口

3.1 添加远程调用方法

说明：因为我这里的设计是在商品上架的时候，向MQ发送的是SKU主键ID，并不是该sku的所有基本信息，所以我需要额外写接口来根据sku主键id查询基本信息，这里大家根据自身情况来写即可。

找到我们的商品模块，先创建一个api包，然后将提供给远程调用的接口放在api包下

/**
 * 远程调用API（生产者）
 * @author Eric
 * @date 2023-06-29 10:00
 */
@RestController
@RequestMapping("/api/product")
public class ProductInnnerController {

    @Autowired
    private CategoryService categoryService;
    @Autowired
    private SkuInfoService skuInfoService;

    @ApiOperation(value ="根据分类id获取分类信息")
    @GetMapping("/inner/getCategory/{categoryId}")
    public Category getCategory(@PathVariable Long categoryId){return categoryService.getById(categoryId);}

    @ApiOperation(value ="根据skuId获取sku信息")
    @GetMapping("/  inner/getSkuInfo/{skuId}")
    public SkuInfo getSkuInfo(@PathVariable("skuId") Long skuId){return skuInfoService.getById(skuId);}}

3.2、创建远程调用模块

1、创建service-client模块（我这里的设定是所有远程调用的服务都放在该模块中，但该模块并不负责调用，而是由各自对应的子模块服务来进行调用）
2、service-client模块下再创建子模块 service-product-client定义接口
我的结构如下：

3、service-client模块引入依赖：（主要还是openfeign的依赖）

<dependencies><dependency><groupId>com.atguigu</groupId><artifactId>common-util</artifactId><version>1.0-SNAPSHOT</version><scope>provided</scope></dependency><dependency><groupId>com.atguigu</groupId><artifactId>model</artifactId><version>1.0-SNAPSHOT</version><scope>provided</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><scope>provided</scope></dependency><!-- 服务调用feign --><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-openfeign</artifactId><scope>provided</scope></dependency></dependencies>

4、ProductFeignClient添加定义方法

/**
 * 远程调用其他模块中的api
 * @author Eric
 * @date 2023-06-29 10:04
 */
@FeignClient(value ="service-product") //指定调用模块
public interface ProductFeignClient {

    //根据分类id获取分类信息
    @GetMapping("/api/product/inner/getCategory/{categoryId}")
    public Category getCategory(@PathVariable("categoryId") Long categoryId);

    //根据skuId获取sku信息
    @GetMapping("/api/product/inner/getSkuInfo/{skuId}")
    public SkuInfo getSkuInfo(@PathVariable("skuId") Long skuId);}

3.3、开发service-search 模块接口

1、controller

import com.atguigu.ssyx.common.result.Result;import com.atguigu.ssyx.search.service.SkuService;import io.swagger.annotations.ApiOperation;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.PathVariable;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;

/**
 * 商品搜索列表接口
 *
 * @author Eric
 * @date 2023-06-29 10:15
 */
@RestController
@RequestMapping("api/search/sku")
public class SkuApiController {

    @Autowired
    private SkuService skuService;

    @ApiOperation(value ="上架商品")
    @GetMapping("inner/upperSku/{skuId}")
    public Result upperGoods(@PathVariable("skuId") Long skuId){
        skuService.upperSku(skuId);return Result.ok();}

    @ApiOperation(value ="下架商品")
    @GetMapping("inner/lowerSku/{skuId}")
    public Result lowerGoods(@PathVariable("skuId") Long skuId){
        skuService.lowerSku(skuId);return Result.ok();}}

2、service接口

/**
 * @author Eric
 * @date 2023-06-29 10:16
 */
public interface SkuService {
    /**
     * 上架商品列表
     * @param skuId
     */
    void upperSku(Long skuId);

    /**
     * 下架商品列表
     * @param skuId
     */
    void lowerSku(Long skuId);}

3、impl

import com.alibaba.fastjson.JSON;import com.atguigu.ssyx.enums.SkuType;import com.atguigu.ssyx.model.product.Category;import com.atguigu.ssyx.model.product.SkuInfo;import com.atguigu.ssyx.model.search.SkuEs;import com.atguigu.ssyx.product.client.ProductFeignClient;import com.atguigu.ssyx.search.repository.SkuRepository;import com.atguigu.ssyx.search.service.SkuService;import lombok.extern.slf4j.Slf4j;import org.elasticsearch.client.RestHighLevelClient;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;

/**
 * @author Eric
 * @date 2023-06-29 10:16
 */

@Slf4j
@Service
public class SkuServiceImpl implements SkuService {

    @Autowired
    private ProductFeignClient productFeignClient;

    @Autowired
    private SkuRepository skuEsRepository;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 上架商品列表
     *
     * @param skuId
     */
    @Override
    public void upperSku(Long skuId){
        log.info("upperSku：" + skuId);
        SkuEs skuEs = new SkuEs();

        //查询sku信息
        SkuInfo skuInfo = productFeignClient.getSkuInfo(skuId);if(null == skuInfo)return;

        // 查询分类
        Category category = productFeignClient.getCategory(skuInfo.getCategoryId());if(category != null){
            skuEs.setCategoryId(category.getId());
            skuEs.setCategoryName(category.getName());}
        skuEs.setId(skuInfo.getId());
        skuEs.setKeyword(skuInfo.getSkuName() + "," + skuEs.getCategoryName());
        skuEs.setWareId(skuInfo.getWareId());
        skuEs.setIsNewPerson(skuInfo.getIsNewPerson());
        skuEs.setImgUrl(skuInfo.getImgUrl());
        skuEs.setTitle(skuInfo.getSkuName());if(skuInfo.getSkuType()== SkuType.COMMON.getCode()){
            skuEs.setSkuType(0);
            skuEs.setPrice(skuInfo.getPrice().doubleValue());
            skuEs.setStock(skuInfo.getStock());
            skuEs.setSale(skuInfo.getSale());
            skuEs.setPerLimit(skuInfo.getPerLimit());}else{
            //TODO 待完善-秒杀商品

        }
        SkuEs save = skuEsRepository.save(skuEs);//往Es中新增数据
        log.info("upperSku：" + JSON.toJSONString(save));}

    /**
     * 下架商品列表
     *
     * @param skuId
     */
    @Override
    public void lowerSku(Long skuId){
        this.skuEsRepository.deleteById(skuId);//删除Es中的数据
    }}

4、创建SkuRepository（用来操作Es，这里使用的是SpringData技术）

/**
 * @author Eric
 * @date 2023-06-29 10:19
 */
// 参数一：泛型  参数二：类型，是由泛型中的主键类型而决定的
public interface SkuRepository extends ElasticsearchRepository<SkuEs, Long>{}

4、RabbitMQ

1、创建mq模块

因为mq的作用类似于工具类，并且多处需要使用，所以我这里选择放在common模块下

2、引入MQ依赖

<dependencies><!--rabbitmq消息队列--><dependency><groupId>org.springframework.cloud</groupId><artifactId>spring-cloud-starter-bus-amqp</artifactId></dependency></dependencies>

3、添加service方法

import org.springframework.amqp.AmqpException;import org.springframework.amqp.core.Message;import org.springframework.amqp.core.MessagePostProcessor;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;

/**
 * @author Eric
 * @date 2023-06-30 22:57
 */

@Service
public class RabbitService {

    //  引入操作rabbitmq 的模板
    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 发送消息
     *
     * @param exchange   交换机
     * @param routingKey 路由键（路由key）
     * @param message    消息
     * @return
     */
    public boolean sendMessage(String exchange, String routingKey, Object message){
        //  调用发送数据的方法
        rabbitTemplate.convertAndSend(exchange, routingKey, message);returntrue;}

    /**
     * 发送延迟消息的方法
     *
     * @param exchange   交换机
     * @param routingKey 路由键
     * @param message    消息内容
     * @param delayTime  延迟时间
     * @return
     */
    public boolean sendDelayMessage(String exchange, String routingKey, Object message, int delayTime){

        //  在发送消息的时候设置延迟时间
        rabbitTemplate.convertAndSend(exchange, routingKey, message, new MessagePostProcessor(){
            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                //  设置一个延迟时间
                message.getMessageProperties().setDelay(delayTime * 1000);return message;}});returntrue;}}

4、配置mq消息转换器

import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;import org.springframework.amqp.support.converter.MessageConverter;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;

/**
 * mq消息转换器（默认是字符串转换器）
 */

@Configuration
public class MQConfig {

    @Bean
    public MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}}

5、添加消息的确认配置（我这里配置的是手动确认模式）

import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import javax.annotation.PostConstruct;

@Component
public class MQProducerAckConfig implements RabbitTemplate.ReturnCallback,RabbitTemplate.ConfirmCallback {

    //  我们发送消息使用的是 private RabbitTemplate rabbitTemplate; 对象
    //  如果不做设置的话 当前的rabbitTemplate 与当前的配置类没有任何关系！
    @Autowired
    private RabbitTemplate rabbitTemplate;

    //  设置 表示修饰一个非静态的void方法，在服务器加载Servlet的时候运行。并且只执行一次！
    @PostConstruct
    public void init(){
        rabbitTemplate.setReturnCallback(this);
        rabbitTemplate.setConfirmCallback(this);}

    /**
     * 表示消息是否正确发送到了交换机上
     * @param correlationData   消息的载体
     * @param ack   判断是否发送到交换机上
     * @param cause 原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause){
        if(ack){
            System.out.println("消息发送成功！");}else {
            System.out.println("消息发送失败！"+cause);}}

    /**
     * 消息如果没有正确发送到队列中，则会走这个方法！如果消息被正常处理，则这个方法不会走！
     * @param message
     * @param replyCode
     * @param replyText
     * @param exchange
     * @param routingKey
     */
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey){
        System.out.println("消息主体: " + new String(message.getBody()));
        System.out.println("应答码: " + replyCode);
        System.out.println("描述：" + replyText);
        System.out.println("消息使用的交换器 exchange : " + exchange);
        System.out.println("消息使用的路由键 routing : " + routingKey);}}

6、RabbitMQ常量类

/**
 * 消息队列常量类
 */
public class MqConst {
    /**
     * 消息补偿
     */
    public static final String MQ_KEY_PREFIX ="ssyx.mq:list";
    public static final int RETRY_COUNT =3;

    /**
     * 商品上下架
     */
    public static final String EXCHANGE_GOODS_DIRECT ="ssyx.goods.direct";
    public static final String ROUTING_GOODS_UPPER ="ssyx.goods.upper";
    public static final String ROUTING_GOODS_LOWER ="ssyx.goods.lower";
    //队列
    public static final String QUEUE_GOODS_UPPER ="ssyx.goods.upper";
    public static final String QUEUE_GOODS_LOWER ="ssyx.goods.lower";

    /**
     * 团长上下线
     */
    public static final String EXCHANGE_LEADER_DIRECT ="ssyx.leader.direct";
    public static final String ROUTING_LEADER_UPPER ="ssyx.leader.upper";
    public static final String ROUTING_LEADER_LOWER ="ssyx.leader.lower";
    //队列
    public static final String QUEUE_LEADER_UPPER ="ssyx.leader.upper";
    public static final String QUEUE_LEADER_LOWER ="ssyx.leader.lower";

    //订单
    public static final String EXCHANGE_ORDER_DIRECT ="ssyx.order.direct";
    public static final String ROUTING_ROLLBACK_STOCK ="ssyx.rollback.stock";
    public static final String ROUTING_MINUS_STOCK ="ssyx.minus.stock";

    public static final String ROUTING_DELETE_CART ="ssyx.delete.cart";
    //解锁普通商品库存
    public static final String QUEUE_ROLLBACK_STOCK ="ssyx.rollback.stock";
    public static final String QUEUE_SECKILL_ROLLBACK_STOCK ="ssyx.seckill.rollback.stock";
    public static final String QUEUE_MINUS_STOCK ="ssyx.minus.stock";
    public static final String QUEUE_DELETE_CART ="ssyx.delete.cart";

    //支付
    public static final String EXCHANGE_PAY_DIRECT ="ssyx.pay.direct";
    public static final String ROUTING_PAY_SUCCESS ="ssyx.pay.success";
    public static final String QUEUE_ORDER_PAY ="ssyx.order.pay";
    public static final String QUEUE_LEADER_BILL ="ssyx.leader.bill";

    //取消订单
    public static final String EXCHANGE_CANCEL_ORDER_DIRECT ="ssyx.cancel.order.direct";
    public static final String ROUTING_CANCEL_ORDER ="ssyx.cancel.order";
    //延迟取消订单队列
    public static final String QUEUE_CANCEL_ORDER ="ssyx.cancel.order";

    /**
     * 定时任务
     */
    public static final String EXCHANGE_DIRECT_TASK ="ssyx.exchange.direct.task";
    public static final String ROUTING_TASK_23 ="ssyx.task.23";
    //队列
    public static final String QUEUE_TASK_23 ="ssyx.queue.task.23";}

5、完善SKU管理商品上下架

5.1、商品服务

1、找到自己的商品服务，引入rabbit-util模块

<!-- 引入MQ服务 --><dependency><groupId>com.atguigu</groupId><artifactId>rabbit_util</artifactId><version>1.0-SNAPSHOT</version></dependency>

2、同时在配置文件添加MQ配置

  rabbitmq:
    host: 192.168.56.101
    port: 5672
    username: guest
    password: guest
    publisher-confirm-type: CORRELATED  #发布确认模式，消息是否被成功发送到交换机
    publisher-returns: true
    listener:
      simple:
        prefetch: 1
        concurrency: 3
        acknowledge-mode: manual   #消费端手动确认

3、修改SkuInfoServiceImpl的publish方法（上下架方法）

@Autowired
private RabbitService rabbitService;

@Transactional(rollbackFor ={Exception.class})
@Override
public void publish(Long skuId, Integer status){
    // 更改发布状态
    if(status ==1){
        SkuInfo skuInfoUp = new SkuInfo();
        skuInfoUp.setId(skuId);
        skuInfoUp.setPublishStatus(1);
        skuInfoMapper.updateById(skuInfoUp);

        //商品上架：发送mq消息同步es
        rabbitService.sendMessage(MqConst.EXCHANGE_GOODS_DIRECT, MqConst.ROUTING_GOODS_UPPER, skuId);}else{
        SkuInfo skuInfoUp = new SkuInfo();
        skuInfoUp.setId(skuId);
        skuInfoUp.setPublishStatus(0);
        skuInfoMapper.updateById(skuInfoUp);

        //商品下架：发送mq消息同步es
        rabbitService.sendMessage(MqConst.EXCHANGE_GOODS_DIRECT, MqConst.ROUTING_GOODS_LOWER, skuId);}}

5.2、es服务

在service-search服务也引入消息依赖

<!-- 引入mq消息服务 --><dependency><groupId>com.atguigu</groupId><artifactId>rabbit_util</artifactId><version>1.0-SNAPSHOT</version></dependency>

2、添加SkuReceiver接收MQ消息方法

import com.atguigu.ssyx.rabbit.constant.MqConst;import com.atguigu.ssyx.search.service.SkuService;import com.rabbitmq.client.Channel;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.Exchange;import org.springframework.amqp.rabbit.annotation.Queue;import org.springframework.amqp.rabbit.annotation.QueueBinding;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.io.IOException;

@Component
public class SkuReceiver {

    @Autowired
    private SkuService skuService;

    /**
     * 商品上架
     * RabbitListener  自动监听
     *
     * @param skuId
     * @param message
     * @param channel
     * @throws IOException
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = MqConst.QUEUE_GOODS_UPPER, durable ="true"),
            exchange = @Exchange(value = MqConst.EXCHANGE_GOODS_DIRECT),
            key ={MqConst.ROUTING_GOODS_UPPER}))
    public void upperSku(Long skuId, Message message, Channel channel) throws IOException {
        System.out.println("上架：消息消费成功！");if(null != skuId){
            skuService.upperSku(skuId);}
        /**
         * 第一个参数：表示收到的消息的标号
         * 第二个参数：如果为true表示可以签收多个消息
         */
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}

    /**
     * 商品下架
     *
     * @param skuId
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = MqConst.QUEUE_GOODS_LOWER, durable ="true"),
            exchange = @Exchange(value = MqConst.EXCHANGE_GOODS_DIRECT),
            key ={MqConst.ROUTING_GOODS_LOWER}))
    public void lowerSku(Long skuId, Message message, Channel channel) throws IOException {
        System.out.println("下架：消息消费成功！");if(null != skuId){
            skuService.lowerSku(skuId);}
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);}}

6、最终测试

测试：需要启动如下服务：
1、Nginx
2、Naocs
3、ElaticSearch
4、kibana
5、RabbitMQ
6、相关微服务模块
7、前端服务

我们先使用Kibana查询Es中的数据

GET /_cat/indices?v

POST /skues/_search
{"query":{"match_all":{}}}

发现此时为空

此时我们任意选择一个SKU点击上架，上架成功，此时我们去到MQ中查看消息，发现消息发送成功

注意，我这里能看到消息是因为我在消费前断点了，如果没有断点的话消息是会被立刻消费掉的，
此刻，我们再去es中查看，发现sku数据成功新增

此时，我们再点击下架。
再重新查询，发现es中的sku数据成功删除！

总结

写完我们会发现，整个过程其实并不难，总的调用过程大概是：后台服务选择SKU，点击上架 -》此时调用商品服务 -》然后商品服务发送消息给RabbitMQ -》而我们的ES服务在监听着 -》收到消息后 立刻进行消费，同时操作ES
整个过程如果用一张图来解释的话，大概如下：