商品和订单中使用MQ

AlgerFan · Aug 27, 2019 · b4bcffd · b4bcffd
1 parent 4a84e6c
commit b4bcffd
Show file tree

Hide file tree

Showing 8 changed files with 179 additions and 21 deletions.
diff --git a/order/order-server/pom.xml b/order/order-server/pom.xml
@@ -21,6 +21,10 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-openfeign</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-config-client</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-bus-amqp</artifactId>
@@ -29,6 +33,10 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-stream-rabbit</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
         <!--     TODO     spring-cloud-starter-hystrix ——> spring-cloud-starter-netflix-hystrix-->
         <!--     TODO   2.0.2.RELEASE需要引入-->
         <dependency>
@@ -51,10 +59,6 @@
             <groupId>com.google.code.gson</groupId>
             <artifactId>gson</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.springframework.cloud</groupId>
-            <artifactId>spring-cloud-config-client</artifactId>
-        </dependency>
         <dependency>
             <groupId>cn.algerfan</groupId>
             <artifactId>product-client</artifactId>

diff --git a/order/order-server/src/main/java/cn/algerfan/order/server/message/ProductReceiver.java b/order/order-server/src/main/java/cn/algerfan/order/server/message/ProductReceiver.java
@@ -0,0 +1,42 @@
+package cn.algerfan.order.server.message;
+
+import cn.algerfan.order.server.util.JsonUtil;
+import cn.algerfan.product.common.ProductInfoOutput;
+import com.fasterxml.jackson.core.type.TypeReference;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * @author algerfan
+ * @time 2019/8/27 11:59
+ */
+@Component
+@Slf4j
+public class ProductReceiver {
+
+    private static final String PRODUCT_STOCK_TEMPLATE = "product_stock_%s";
+
+    @Autowired
+    private StringRedisTemplate stringRedisTemplate;
+
+    @RabbitListener(queuesToDeclare = @Queue("productInfo"))
+    public void process(String message) {
+        //message——>productInfoOutput
+        List<ProductInfoOutput> productInfoOutputList = (List<ProductInfoOutput>) JsonUtil.fromJson(message,
+                new TypeReference<List<ProductInfoOutput>>() {});
+        log.info("从队列{}接收到消息{}", "productInfo", productInfoOutputList);
+        //存储在redis中
+        assert productInfoOutputList != null;
+        for (ProductInfoOutput productInfoOutput : productInfoOutputList) {
+            stringRedisTemplate.opsForValue().set(String.format(PRODUCT_STOCK_TEMPLATE, productInfoOutput.getProductId()),
+                    String.valueOf(productInfoOutput.getProductStock()));
+        }
+    }
+
+}
diff --git a/order/order-server/src/main/java/cn/algerfan/order/server/util/JsonUtil.java b/order/order-server/src/main/java/cn/algerfan/order/server/util/JsonUtil.java
@@ -0,0 +1,60 @@
+package cn.algerfan.order.server.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+import java.io.IOException;
+
+/**
+ * @author algerfan
+ */
+public class JsonUtil {
+
+	private static ObjectMapper objectMapper = new ObjectMapper();
+
+	/**
+	 * 转换为json字符串
+	 * @param object
+	 * @return
+	 */
+	public static String toJson(Object object) {
+		try {
+			return objectMapper.writeValueAsString(object);
+		} catch (JsonProcessingException e) {
+			e.printStackTrace();
+		}
+		return null;
+	}
+
+	/**
+	 * json转对象
+	 * @param string
+	 * @param classType
+	 * @return
+	 */
+	public static Object fromJson(String string, Class classType) {
+		try {
+			return objectMapper.readValue(string, classType);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+		return null;
+	}
+
+
+	/**
+	 * json转对象
+	 * @param string
+	 * @param typeReference
+	 * @return
+	 */
+	public static Object fromJson(String string, TypeReference typeReference) {
+		try {
+			return objectMapper.readValue(string, typeReference);
+		} catch (IOException e) {
+			e.printStackTrace();
+		}
+		return null;
+	}
+}
diff --git a/product/product-server/pom.xml b/product/product-server/pom.xml
@@ -20,6 +20,14 @@
             <groupId>org.springframework.cloud</groupId>
             <artifactId>spring-cloud-starter-openfeign</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-config-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-bus-amqp</artifactId>
+        </dependency>
         <!--     TODO   2.0.2.RELEASE需要引入-->
         <dependency>
             <groupId>org.springframework.boot</groupId>

diff --git a/...duct-server/src/main/java/cn/algerfan/product/server/service/impl/ProductServiceImpl.java b/...duct-server/src/main/java/cn/algerfan/product/server/service/impl/ProductServiceImpl.java
@@ -8,6 +8,8 @@
 import cn.algerfan.product.server.exception.ProductException;
 import cn.algerfan.product.server.repository.ProductInfoRepository;
 import cn.algerfan.product.server.service.ProductService;
+import cn.algerfan.product.server.util.JsonUtil;
+import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
@@ -16,6 +18,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
+import java.util.stream.Collectors;
 
 /**
  * @author algerfan
@@ -26,6 +29,8 @@ public class ProductServiceImpl implements ProductService {
 
     @Autowired
     private ProductInfoRepository productInfoRepository;
+    @Autowired
+    private AmqpTemplate amqpTemplate;
 
     @Override
     public List<ProductInfo> findUpAll() {
@@ -45,8 +50,20 @@ public List<ProductInfoOutput> productInfoList(List<String> productIdList) {
     }
 
     @Override
-    @Transactional(rollbackFor = Exception.class)
     public void decreaseStock(List<DecreaseStockOutput> cartDtoList) {
+        List<ProductInfo> productInfoList = decreaseStockProcess(cartDtoList);
+        //发送mq消息
+        List<ProductInfoOutput> collect = productInfoList.stream().map(e -> {
+            ProductInfoOutput productInfoOutput = new ProductInfoOutput();
+            BeanUtils.copyProperties(e, productInfoOutput);
+            return productInfoOutput;
+        }).collect(Collectors.toList());
+        amqpTemplate.convertAndSend("productInfo", JsonUtil.toJson(collect));
+    }
+
+    @Transactional(rollbackFor = Exception.class)
+    public List<ProductInfo> decreaseStockProcess(List<DecreaseStockOutput> cartDtoList) {
+        List<ProductInfo> productInfoList = new ArrayList<>();
         for (DecreaseStockOutput cartDto : cartDtoList) {
             Optional<ProductInfo> productInfoOptional = productInfoRepository.findById(cartDto.getProductId());
             //判断商品是否存在
@@ -61,6 +78,8 @@ public void decreaseStock(List<DecreaseStockOutput> cartDtoList) {
             }
             productInfo.setProductStock(result);
             productInfoRepository.save(productInfo);
+            productInfoList.add(productInfo);
         }
+        return productInfoList;
     }
 }
diff --git a/product/product-server/src/main/java/cn/algerfan/product/server/util/JsonUtil.java b/product/product-server/src/main/java/cn/algerfan/product/server/util/JsonUtil.java
@@ -0,0 +1,26 @@
+package cn.algerfan.product.server.util;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+/**
+ * @author algerfan
+ */
+public class JsonUtil {
+
+	private static ObjectMapper objectMapper = new ObjectMapper();
+
+	/**
+	 * 转换为json字符串
+	 * @param object
+	 * @return
+	 */
+	public static String toJson(Object object) {
+		try {
+			return objectMapper.writeValueAsString(object);
+		} catch (JsonProcessingException e) {
+			e.printStackTrace();
+		}
+		return null;
+	}
+}
diff --git a/product/product-server/src/main/resources/application.yml b/product/product-server/src/main/resources/application.yml
diff --git a/product/product-server/src/main/resources/bootstrap.yml b/product/product-server/src/main/resources/bootstrap.yml
@@ -0,0 +1,15 @@
+eureka:
+  client:
+    service-url:
+      defaultZone: http://127.0.0.1:8761/eureka/
+  instance:
+    prefer-ip-address: true
+spring:
+  application:
+    name: product
+  cloud:
+    config:
+      discovery:
+        enabled: true
+        service-id: CONFIG
+      profile: dev