更新

2026-10-28 16:13:43 +00:00 · 2019-05-22 23:26:14 +08:00
parent 9ae00ac55d
commit 7bfcd1d288
1 changed files with 96 additions and 0 deletions
@@ -0,0 +1,96 @@
+package org.itkk.udf.demo.server;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+@Slf4j
+public class DbQueue {
+    private final AtomicBoolean executionFlag;
+    private final String queueName;
+    private final LinkedBlockingQueue<String> queue;
+    private final ExecutorService executorService;
+
+    public DbQueue(String queueName, int concurrentCount, int maxQueueDepth) {
+        this.executionFlag = new AtomicBoolean();
+        this.queueName = queueName;
+        this.queue = new LinkedBlockingQueue<>(maxQueueDepth);
+        this.executorService = Executors.newFixedThreadPool(concurrentCount);
+    }
+
+    public void put(String s) {
+        //非阻塞的添加到队列,可能队里已经满了,添加不成功
+        this.queue.offer(s);
+        //将任务添加到数据库
+        //todo insert to db
+    }
+
+    private String get() {
+        String id = this.queue.poll();
+        if (StringUtils.isBlank(id)) {
+            List<String> ids = new ArrayList<>();
+            if (CollectionUtils.isNotEmpty(ids)) {
+                for (String idTemp : ids) {
+                    boolean addFlag = this.queue.offer(idTemp);
+                    if (!addFlag) {
+                        break;
+                    }
+                }
+            }
+            try {
+                id = this.queue.take();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+        return id;
+    }
+
+    private void execute() {
+        while (true) {
+            String id = this.get();
+            this.executorService.execute(() -> {
+                //[锁定]并且[变更任务状态] - 开启事务
+                try {
+                    try {
+                        Object task = id; //TODO 从数据库里获得任务对象,状态为[待执行] for update nowait
+                    } catch (Exception e) {
+                        log.error("锁定任务失败", e);
+                        //[锁定]并且[变更任务状态] - 回滚事务
+                        return;
+                    }
+                    Object task = "update status"; //todo 更新数据库状态为[执行中]
+                    //[锁定]并且[变更任务状态] - 提交事务
+                } catch (Exception e) {
+                    log.error("任务状态更新失败", e);
+                    //[锁定]并且[变更任务状态] - 回滚事务
+                    return;
+                }
+                //执行任务
+                boolean execFlag = false;
+                try {
+                    System.out.println(id);
+                    execFlag = true;
+                } catch (Exception e) {
+                    log.error("任务执行失败", e);
+                    execFlag = false;
+                }
+                //最后更新任务状态 - 开启事务
+                try{
+
+                }finally {
+
+                }
+                //最后更新任务状态 - 提交事务
+            });
+        }
+    }
+
+}