[feature]添加重试以及多线程并发策略

boyunkai · boyunkai · commit d44d674e92ff · 2021-02-07T17:42:07.000+08:00
diff --git a/09mq/activemq-demo/src/main/java/io/order/QueueApplication.java b/09mq/activemq-demo/src/main/java/io/order/QueueApplication.java
@@ -1,9 +1,5 @@
 package io.order;
 
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.locks.ReentrantLock;
-
 import javax.annotation.Resource;
 
 import org.springframework.boot.ApplicationArguments;
@@ -35,7 +31,7 @@ public static void main(String[] args) {
     }
 
     @Override
-    public void run(ApplicationArguments args) throws Exception {
+    public void run(ApplicationArguments args) {
         for (int i = 0; i < 100; i++) {
             log.info(queueProducer.sendMessage(String.valueOf(i)) + "<=======入队");
         }
diff --git a/09mq/activemq-demo/src/main/java/io/order/model/Order.java b/09mq/activemq-demo/src/main/java/io/order/model/Order.java
@@ -4,13 +4,11 @@
 import lombok.Data;
 
 /**
+ * 订单表
+ *
  * @author boyunkai <boyunkai@kuaishou.com>
  * Created on 2021-02-07
  */
-
-/**
- * 订单表
- */
 @Data
 @AllArgsConstructor
 public class Order {
diff --git a/09mq/activemq-demo/src/main/java/io/order/service/QueueConsumer.java b/09mq/activemq-demo/src/main/java/io/order/service/QueueConsumer.java
@@ -4,7 +4,6 @@
 import java.util.Objects;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.locks.ReentrantLock;
 
 import javax.annotation.Resource;
 
@@ -42,7 +41,10 @@ public void receiveMessage() {
                 executorService.submit(() -> {
                     try {
                         // 重试策略
-                        retryBuss(order);
+                        int success = retryBuss(order);
+                        if (success <= 0) {
+                            log.info("订单" + order.getOrderId() + "=======>出队失败");
+                        }
                     } catch (InterruptedException exception) {
                         log.info("进程被打断");
                         Thread.currentThread().interrupt();
@@ -52,14 +54,14 @@ public void receiveMessage() {
         }
     }
 
-    private void retryBuss(Order order) throws InterruptedException {
+    private int retryBuss(Order order) throws InterruptedException {
         int retryNum = 1;
         while (retryNum <= TRY_TIMES) {
             try {
                 // 加锁解决并发，效率降低
                 int success = updateOrder(order);
                 if (success > 0L) {
-                    break;
+                    return 1;
                 }
                 retryNum++;
             } catch (Exception e) {
@@ -68,6 +70,7 @@ private void retryBuss(Order order) throws InterruptedException {
                 continue;
             }
         }
+        return 0;
     }
 
     private synchronized int updateOrder(Order order) throws IllegalAccessException, InterruptedException {
diff --git a/09mq/activemq-demo/src/main/java/io/order/service/QueueProducer.java b/09mq/activemq-demo/src/main/java/io/order/service/QueueProducer.java
@@ -1,7 +1,5 @@
 package io.order.service;
 
-import java.util.UUID;
-
 import javax.annotation.Resource;
 
 import org.springframework.stereotype.Service;
@@ -22,7 +20,7 @@ public class QueueProducer {
     @Resource
     OrderMapper orderMapper;
 
-    public String sendMessage(String orderId) throws IllegalAccessException {
+    public String sendMessage(String orderId) {
         // STEP 1: 生成订单
         Order order = new Order(0, orderId, 0);
         // STEP 2: 发送订单消息
