😊 @ 作者: 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、前置条件
- Nacos的运行
- ElasticSearch的运行(我这里使用的是IK分词器和Kibana工具)
- RabbitMQ的运行
- 项目使用的是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);
return true;
}
/**
* 发送延迟消息的方法
*
* @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;
}
});
return true;
}
}
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数据成功删除!
文章来源:https://www.toymoban.com/news/detail-665929.html
总结
写完我们会发现,整个过程其实并不难,总的调用过程大概是:后台服务选择SKU,点击上架 -》此时调用商品服务 -》然后商品服务发送消息给RabbitMQ -》而我们的ES服务在监听着 -》收到消息后 立刻进行消费,同时操作ES
整个过程如果用一张图来解释的话,大概如下:
文章来源地址https://www.toymoban.com/news/detail-665929.html
到了这里,关于SpringBoot实战项目整合RabbitMQ+ElaticSearch实现SKU上下架功能的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!