feat(search): 整合rabbitMQ发消息，商品上下架

.idea/encodings.xml

@@ -4,6 +4,7 @@
     <file url="file://$PROJECT_DIR$/common/common-util/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/common/common-util/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/common/common-util/src/main/resources-filtered" charset="UTF-8" />
+    <file url="file://$PROJECT_DIR$/common/rabbit-util/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/common/service-util/src/main/java" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/common/service-util/src/main/resources" charset="UTF-8" />
     <file url="file://$PROJECT_DIR$/common/service-util/src/main/resources-filtered" charset="UTF-8" />

common/pom.xml

@@ -15,6 +15,7 @@
     <modules>
         <module>common-util</module>
         <module>service-util</module>
+        <module>rabbit-util</module>
     </modules>
 
     <properties>

common/rabbit-util/pom.xml

@@ -0,0 +1,26 @@
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.atguigu</groupId>
+        <artifactId>common</artifactId>
+        <version>1.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>rabbit-util</artifactId>
+    <name>Archetype - rabbit-util</name>
+    <url>https://maven.apache.org</url>
+
+    <dependencies>
+        <!--rabbitmq消息队列-->
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-bus-amqp</artifactId>
+        </dependency>
+        <!-- 消息转换器 -->
+        <dependency>
+            <groupId>com.fasterxml.jackson.dataformat</groupId>
+            <artifactId>jackson-dataformat-xml</artifactId>
+            <version>2.16.0-rc1</version>
+        </dependency>
+    </dependencies>
+</project>

common/rabbit-util/src/main/java/com/atguigu/ssyx/mq/config/RabbitMQConfiguration.java

@@ -0,0 +1,28 @@
+package com.atguigu.ssyx.mq.config;
+
+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;
+
+@Configuration
+public class RabbitMQConfiguration {
+    /**
+     * * 重写，并引入消息转换器
+     * <dependency>
+     * <groupId>com.fasterxml.jackson.dataformat</groupId>
+     * <artifactId>jackson-dataformat-xml</artifactId>
+     * <version>2.16.0-rc1</version>
+     * </dependency>
+     *
+     * @return MessageConverter
+     */
+    @Bean
+    public MessageConverter jsonMessageConverter() {
+        // 1.定义消息转换器
+        Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
+        // 2.配置自动创建消息id，用于识别不同消息，也可以在业务中基于ID判断是否是重复消息
+        converter.setCreateMessageIds(true);
+        return converter;
+    }
+}

common/rabbit-util/src/main/java/com/atguigu/ssyx/mq/constant/MqConst.java

@@ -0,0 +1,53 @@
+package com.atguigu.ssyx.mq.constant;
+
+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";
+}

common/rabbit-util/src/main/java/com/atguigu/ssyx/mq/service/RabbitService.java

@@ -0,0 +1,22 @@
+package com.atguigu.ssyx.mq.service;
+
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Service
+public class RabbitService {
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    /**
+     * 发送消息
+     *
+     * @param exchange  交换机
+     * @param routerKey 路由
+     * @param message   消息
+     */
+    public void sendMessage(String exchange, String routerKey, Object message) {
+        rabbitTemplate.convertAndSend(exchange, routerKey, message);
+    }
+}

service/service-product/pom.xml

@@ -18,6 +18,12 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>com.atguigu</groupId>
+            <artifactId>rabbit-util</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
+        
         <dependency>
             <groupId>cn.hutool</groupId>
             <artifactId>hutool-all</artifactId>

service/service-product/src/main/java/com/atguigu/ssyx/product/ServiceProductApplication.java

@@ -6,7 +6,7 @@ import org.springframework.context.annotation.ComponentScan;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
 
 @SpringBootApplication
-@ComponentScan(basePackages = {"com.atguigu.ssyx.common"})
+@ComponentScan(basePackages = {"com.atguigu.ssyx.common", "com.atguigu.ssyx.mq"})
 @EnableTransactionManagement
 public class ServiceProductApplication {
     public static void main(String[] args) {

service/service-product/src/main/java/com/atguigu/ssyx/product/service/impl/SkuInfoServiceImpl.java

@@ -4,6 +4,8 @@ import com.atguigu.ssyx.model.product.SkuAttrValue;
 import com.atguigu.ssyx.model.product.SkuImage;
 import com.atguigu.ssyx.model.product.SkuInfo;
 import com.atguigu.ssyx.model.product.SkuPoster;
+import com.atguigu.ssyx.mq.constant.MqConst;
+import com.atguigu.ssyx.mq.service.RabbitService;
 import com.atguigu.ssyx.product.mapper.SkuInfoMapper;
 import com.atguigu.ssyx.product.service.SkuAttrValueService;
 import com.atguigu.ssyx.product.service.SkuImageService;
@@ -41,7 +43,8 @@ public class SkuInfoServiceImpl extends ServiceImpl<SkuInfoMapper, SkuInfo> impl
     private SkuImageService skuImageService;
     @Autowired
     private SkuPosterService skuPosterService;
-
+    @Autowired
+    private RabbitService rabbitService;
 
     /**
      * 获取sku分页列表
@@ -197,8 +200,12 @@ public class SkuInfoServiceImpl extends ServiceImpl<SkuInfoMapper, SkuInfo> impl
         skuInfo.setId(skuId);
         if (status == 1) {
             skuInfo.setPublishStatus(status);
+            // 商品上架，发送MQ消息
+            rabbitService.sendMessage(MqConst.EXCHANGE_GOODS_DIRECT, MqConst.ROUTING_GOODS_UPPER, skuId);
         } else {
             skuInfo.setPublishStatus(0);
+            // 商品下架：发送mq消息同步es
+            rabbitService.sendMessage(MqConst.EXCHANGE_GOODS_DIRECT, MqConst.ROUTING_GOODS_LOWER, skuId);
         }
         baseMapper.updateById(skuInfo);
     }

service/service-product/src/main/resources/application-dev.yml

@@ -2,6 +2,12 @@ server:
   port: 8203
 
 bunny:
+  rabbitmq:
+    host: 192.168.1.4
+    port: 5672
+    username: bunny
+    password: "02120212"
+
   datasource:
     host: 106.15.251.123
     port: 3305

service/service-product/src/main/resources/application.yml

@@ -26,6 +26,24 @@ spring:
     date-format: yyyy-MM-dd HH:mm:ss
     time-zone: GMT+8
 
+  rabbitmq:
+    host: ${bunny.rabbitmq.host}
+    port: ${bunny.rabbitmq.port}
+    username: ${bunny.rabbitmq.username}
+    password: ${bunny.rabbitmq.password}
+    publisher-confirm-type: CORRELATED
+    publisher-returns: true
+    listener:
+      simple:
+        prefetch: 1
+        concurrency: 3
+        acknowledge-mode: manual
+        retry:
+          enabled: true # 开启消费者失败重试
+          initial-interval: 1000ms # 初始失败等待时长
+          multiplier: 1 # 下次失败等待时间被树，下次等待时长 multiplier * last-interval
+          max-attempts: 3 # 最大重试次数
+          stateless: true # true 无状态 false 有状态。如果业务中包含事务，这里改为false
 
 mybatis-plus:
   type-aliases-package: com.atguigu.model # 配置每个包前缀

service/service-search/pom.xml

@@ -18,6 +18,11 @@
     </properties>
 
     <dependencies>
+        <dependency>
+            <groupId>com.atguigu</groupId>
+            <artifactId>rabbit-util</artifactId>
+            <version>1.0-SNAPSHOT</version>
+        </dependency>
         <dependency>
             <groupId>com.atguigu</groupId>
             <artifactId>service-product-client</artifactId>

service/service-search/src/main/java/com/atguigu/ssyx/search/MqListener/SkuListener.java

@@ -0,0 +1,40 @@
+package com.atguigu.ssyx.search.MqListener;
+
+import com.atguigu.ssyx.mq.constant.MqConst;
+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.stereotype.Component;
+
+import java.io.IOException;
+
+@Component
+public class SkuListener {
+    // 商品上架
+    @RabbitListener(bindings = @QueueBinding(
+            value = @Queue(name = MqConst.QUEUE_GOODS_UPPER, durable = "true"),
+            exchange = @Exchange(name = MqConst.EXCHANGE_GOODS_DIRECT),
+            key = {MqConst.ROUTING_GOODS_UPPER}
+    ))
+    public void upperSku(Long skuId, Message message, Channel channel) throws IOException {
+        if (skuId != null) {
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+        }
+    }
+
+    // 商品下架
+    @RabbitListener(bindings = @QueueBinding(
+            value = @Queue(name = MqConst.QUEUE_GOODS_LOWER, durable = "true"),
+            exchange = @Exchange(name = MqConst.EXCHANGE_GOODS_DIRECT),
+            key = {MqConst.ROUTING_GOODS_LOWER}
+    ))
+    public void lowerSku(Long skuId, Message message, Channel channel) throws IOException {
+        if (skuId != null) {
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+        }
+    }
+}
+

service/service-search/src/main/java/com/atguigu/ssyx/search/ServiceSearchApplication.java

@@ -8,7 +8,7 @@ import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.annotation.ComponentScan;
 
 @SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
-@ComponentScan(basePackages = {"com.atguigu.ssyx.common"})
+@ComponentScan(basePackages = {"com.atguigu.ssyx.common", "com.atguigu.ssyx.mq"})
 @EnableFeignClients(basePackages = {"com.atguigu.ssyx.client"})
 @EnableDiscoveryClient
 public class ServiceSearchApplication {

service/service-search/src/main/resources/application-dev.yml

@@ -3,7 +3,7 @@ server:
 
 bunny:
   rabbitmq:
-    host: 116.196.101.14
+    host: 192.168.1.4
     port: 5672
     username: bunny
     password: "02120212"

service/service-search/src/main/resources/application.yml

@@ -34,6 +34,12 @@ spring:
         prefetch: 1
         concurrency: 3
         acknowledge-mode: manual
+        retry:
+          enabled: true # 开启消费者失败重试
+          initial-interval: 1000ms # 初始失败等待时长
+          multiplier: 1 # 下次失败等待时间被树，下次等待时长 multiplier * last-interval
+          max-attempts: 3 # 最大重试次数
+          stateless: true # true 无状态 false 有状态。如果业务中包含事务，这里改为false
 
   cloud:
     sentinel:
@@ -60,7 +66,7 @@ feign:
     enabled: true
   client:
     config:
-      default:   #配置全局的feign的调用超时时间  如果 有指定的服务配置 默认的配置不会生效
+      default: #配置全局的feign的调用超时时间  如果 有指定的服务配置 默认的配置不会生效
         connectTimeout: 30000 # 指定的是 消费者 连接服务提供者的连接超时时间 是否能连接  单位是毫秒
         readTimeout: 50000  # 指定的是调用服务提供者的 服务 的超时时间（）  单位是毫秒
 

service/service-sys/src/main/resources/application-dev.yml

@@ -13,9 +13,3 @@ bunny:
     server-addr: z-bunny.cn:8848
     discovery:
       namespace: ssyx
-
-  minio:
-    endpointUrl: "http://129.211.31.58:9000"
-    bucket-name: ssyx
-    accessKey: bunny
-    secretKey: "02120212"

service/service-sys/src/main/resources/application.yml

@@ -54,10 +54,3 @@ logging:
     dateformat: HH:mm:ss:SSS
   file:
     path: "logs/${spring.application.name}"
-    
-#  bunny:
-#    minio:
-#      endpointUrl: ${bunny.minio.endpointUrl}
-#      accessKey: ${bunny.minio.accessKey}
-#      secretKey: ${bunny.minio.secretKey}
-#      bucket-name: ${bunny.minio.bucket-name}