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

boyunkai · boyunkai · commit 0ce4b5abae58 · 2021-02-07T17:33:43.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,5 +1,9 @@
 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;
@@ -32,19 +36,13 @@ public static void main(String[] args) {
 
     @Override
     public void run(ApplicationArguments args) throws Exception {
-        for (int i = 1; i <= 10; i++) {
-            log.info(queueProducer.sendMessage() + "<=======入队");
+        for (int i = 0; i < 100; i++) {
+            log.info(queueProducer.sendMessage(String.valueOf(i)) + "<=======入队");
         }
-        while (true) {
-            try {
-                queueConsumer.receiveMessage();
-                Thread.sleep(100);
-            } catch (IllegalAccessException exception) {
-                log.info(exception.getMessage());
-            } catch (IllegalStateException exception){
-                log.info(exception.getMessage());
-                break;
-            }
+        try {
+            queueConsumer.receiveMessage();
+        } catch (IllegalStateException exception) {
+            log.info(exception.getMessage());
         }
     }
 }
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
@@ -2,6 +2,9 @@
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.locks.ReentrantLock;
 
 import javax.annotation.Resource;
 
@@ -22,29 +25,67 @@
 public class QueueConsumer {
     @Resource
     OrderMapper orderMapper;
+    private static final ExecutorService executorService = Executors.newFixedThreadPool(3);
+    //最大重试次数
+    private static final Integer TRY_TIMES = 6;
+    //重试间隔时间单位秒
+    private static final Long INTERVAL_TIME = 100L;
 
-    public void receiveMessage() throws IllegalAccessException {
+    public void receiveMessage() {
         // STEP 1: 获取为处理订单列表
         List<Order> orderList = orderMapper.selectByStatusId(0);
         if (Objects.isNull(orderList) || orderList.isEmpty()) {
             throw new IllegalStateException("所有订单已处理");
         }
         for (Order order : orderList) {
-            // STEP 2: 校验订单
-            if (Objects.isNull(order)) {
-                throw new IllegalAccessException("订单不存在");
+            for (int i = 0; i < 10; i++) {
+                executorService.submit(() -> {
+                    try {
+                        // 重试策略
+                        retryBuss(order);
+                    } catch (InterruptedException exception) {
+                        log.info("进程被打断");
+                        Thread.currentThread().interrupt();
+                    }
+                });
             }
-            boolean isOrderComplete = order.getStatusId() == 1;
-            if (isOrderComplete) {
-                throw new IllegalAccessException("订单已完成");
-            }
-            // STEP 3: 插入订单
-            order.setStatusId(1);
-            int success = orderMapper.updateByPrimaryKeySelective(order);
-            if (success <= 0L) {
-                throw new IllegalAccessException("更新订单失败");
+        }
+    }
+
+    private void retryBuss(Order order) throws InterruptedException {
+        int retryNum = 1;
+        while (retryNum <= TRY_TIMES) {
+            try {
+                // 加锁解决并发，效率降低
+                int success = updateOrder(order);
+                if (success > 0L) {
+                    break;
+                }
+                retryNum++;
+            } catch (Exception e) {
+                retryNum++;
+                Thread.sleep(INTERVAL_TIME);
+                continue;
             }
-            log.info("订单" + order.getOrderId() + "=======>出队");
         }
     }
+
+    private synchronized int updateOrder(Order order) throws IllegalAccessException, InterruptedException {
+        // STEP 1: 校验订单
+        if (Objects.isNull(order)) {
+            throw new IllegalAccessException("订单不存在");
+        }
+        if (order.getStatusId() == 1) {
+            return 1;
+        }
+        // STEP 2: 插入订单
+        order.setStatusId(1);
+        int success = orderMapper.updateByPrimaryKeySelective(order);
+        if (success <= 0L) {
+            throw new IllegalAccessException("更新订单失败");
+        }
+        Thread.sleep(100);
+        log.info("订单" + order.getOrderId() + "=======>出队");
+        return success;
+    }
 }
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
@@ -22,9 +22,8 @@ public class QueueProducer {
     @Resource
     OrderMapper orderMapper;
 
-    public String sendMessage() throws IllegalAccessException {
+    public String sendMessage(String orderId) throws IllegalAccessException {
         // STEP 1: 生成订单
-        String orderId = UUID.randomUUID().toString().replace("-", "").substring(0, 10);
         Order order = new Order(0, orderId, 0);
         // STEP 2: 发送订单消息
         int success = orderMapper.insertSelective(order);
