✨ AbstractQueueThread 添加程序关闭时的处理方法.

2021-03-09 11:11:41 +08:00
parent e8e45eadd1
commit e662c4a957
6 changed files with 117 additions and 16 deletions
--- a/ballcat-common/ballcat-common-core/src/main/java/com/hccake/ballcat/common/core/thread/AbstractQueueThread.java
+++ b/ballcat-common/ballcat-common-core/src/main/java/com/hccake/ballcat/common/core/thread/AbstractQueueThread.java
@@ -1,18 +1,23 @@
 package com.hccake.ballcat.common.core.thread;

-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.lang.Nullable;
-
-import javax.validation.constraints.NotNull;
+import com.hccake.ballcat.common.util.JsonUtils;
 import java.util.ArrayList;
 import java.util.List;
+import javax.validation.constraints.NotNull;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.lang.Nullable;

 /**
 * 顶级队列线程类
 *
 * @author lingting 2021/3/2 15:07
 */
-public abstract class AbstractQueueThread<E> extends Thread implements InitializingBean {
+@Slf4j
+public abstract class AbstractQueueThread<E> extends Thread
+		implements InitializingBean, ApplicationListener<ContextClosedEvent> {

 	/**
 	 * 默认缓存数据数量
@@ -133,7 +138,7 @@ public abstract class AbstractQueueThread<E> extends Thread implements Initializ
 						e = poll(getPollTimeoutMs());
 					}
 					catch (InterruptedException interruptedException) {
-						interruptedException.printStackTrace();
+						log.error("{} 类的线程被中断!id: {}", getClass().getSimpleName(), getId());
 					}

 					if (e != null) {
@@ -150,7 +155,13 @@ public abstract class AbstractQueueThread<E> extends Thread implements Initializ
 						break;
 					}
 				}
-				process(list);
+
+				if (!isRun()) {
+					shutdownHandler(list);
+				}
+				else {
+					process(list);
+				}
 			}
 			catch (Throwable e) {
 				error(e, list);
@@ -173,4 +184,35 @@ public abstract class AbstractQueueThread<E> extends Thread implements Initializ
 		start();
 	}

+	@Override
+	public void onApplicationEvent(ContextClosedEvent event) {
+		log.warn("{} 类的线程开始关闭! id: {} ", getClass().getSimpleName(), getId());
+		// 执行关闭方法
+		shutdown();
+	}
+
+	/**
+	 * 线程关闭时执行
+	 * @author lingting 2021-03-08 22:25
+	 */
+	public void shutdown() {
+		// 通过中断线程唤醒当前线程. 让线程进入 shutdownHandler 方法处理数据
+		this.interrupt();
+	}
+
+	/**
+	 * 线程被中断后的处理. 如果有缓存手段可以让数据进入缓存.
+	 * @param list 当前数据
+	 * @author lingting 2021-03-08 22:40
+	 */
+	public void shutdownHandler(List<E> list) {
+		try {
+			log.error("{} 类 线程: {} 被关闭. 数据:{}", this.getClass().getSimpleName(), getId(), JsonUtils.toJson(list));
+		}
+		catch (Throwable e) {
+			log.error("{} 类 线程: {} 被关闭. 数据:{}", this.getClass().getSimpleName(), getId(), list);
+
+		}
+	}
+
 }
--- a/ballcat-extends/ballcat-extend-pay-virtual/src/main/java/com/hccake/starter/pay/viratual/AbstractVerifyThread.java
+++ b/ballcat-extends/ballcat-extend-pay-virtual/src/main/java/com/hccake/starter/pay/viratual/AbstractVerifyThread.java
@@ -66,8 +66,14 @@ public abstract class AbstractVerifyThread<T extends VerifyObj, R> extends Abstr
 	@Override
 	public void receiveProcess(List<T> list, T t) {
 		try {
-			// 收到就处理. 不再汇总处理
-			handler(t, getTransaction(t));
+			if (isRun()) {
+				// 收到就处理. 不再汇总处理. 减少数据丢失的问题
+				handler(t, getTransaction(t));
+			}
+			else {
+				// 结束运行插入redis
+				put(t);
+			}
 		}
 		catch (Exception e) {
 			error(t, e);
--- a/ballcat-samples/ballcat-sample-pay/pom.xml
+++ b/ballcat-samples/ballcat-sample-pay/pom.xml
@@ -47,4 +47,13 @@
            <artifactId>spring-boot-starter-actuator</artifactId>
        </dependency>
    </dependencies>
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <version>${spring-boot.version}</version>
+            </plugin>
+        </plugins>
+    </build>
 </project>
--- a/ballcat-samples/ballcat-sample-pay/src/main/java/com/hccake/sample/pay/virtual/thread/AbstractThread.java
+++ b/ballcat-samples/ballcat-sample-pay/src/main/java/com/hccake/sample/pay/virtual/thread/AbstractThread.java
@@ -24,7 +24,7 @@ public abstract class AbstractThread extends AbstractVerifyThread<Order, Result>

 	@Override
 	public int getBatchSize() {
-		return 1;
+		return 10;
 	}

 	/**
--- a/ballcat-samples/ballcat-sample-pay/src/main/resources/application.yml
+++ b/ballcat-samples/ballcat-sample-pay/src/main/resources/application.yml
@@ -8,14 +8,13 @@ spring:
 ballcat:
  pay:
    bitcoin:
-      omni:
-        endpoints: mainnet
      enabled: true
+      endpoints: mainnet
    ethereum:
      infura:
-        endpoints: mainnet
        # 正式运营请使用自己的 project id, 这个随时可能失效
        project-id: b6066b4cfce54e7384ea38d52f9260ac
+      endpoints: mainnet
    tronscan:
      endpoints: mainnet
      enabled: true
--- a/ballcat-starters/ballcat-spring-boot-starter-redis/src/main/java/com/hccake/ballcat/common/redis/thread/AbstractRedisThread.java
+++ b/ballcat-starters/ballcat-spring-boot-starter-redis/src/main/java/com/hccake/ballcat/common/redis/thread/AbstractRedisThread.java
@@ -6,6 +6,7 @@ import com.hccake.ballcat.common.redis.RedisHelper;
 import com.hccake.ballcat.common.util.JsonUtils;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.util.List;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
@@ -24,15 +25,20 @@ public abstract class AbstractRedisThread<E> extends AbstractQueueThread<E> {
 	@Autowired
 	protected RedisHelper redisHelper;

+	/**
+	 * 是否正在运行
+	 */
+	protected boolean run = true;
+
 	/**
 	 * 锁
 	 */
-	private final ReentrantLock lock = new ReentrantLock();
+	protected final ReentrantLock lock = new ReentrantLock();

 	/**
 	 * 激活与休眠线程
 	 */
-	private final Condition condition = lock.newCondition();
+	protected final Condition condition = lock.newCondition();

 	/**
 	 * 获取数据存储的key
@@ -83,6 +89,7 @@ public abstract class AbstractRedisThread<E> extends AbstractQueueThread<E> {
 			try {
 				lock.lockInterruptibly();
 				try {
+					// 线程被中断后无法执行Redis命令
 					RedisHelper.listRightPush(getKey(), convertToString(e));
 					// 激活线程
 					condition.signal();
@@ -92,7 +99,7 @@ public abstract class AbstractRedisThread<E> extends AbstractQueueThread<E> {
 				}
 			}
 			catch (Exception ex) {
-				log.error("{} put Object error, param: {}", this.getClass().toString(), e, ex);
+				log.error("{} put error, param: {}", this.getClass().toString(), e, ex);
 			}
 		}
 	}
@@ -109,6 +116,10 @@ public abstract class AbstractRedisThread<E> extends AbstractQueueThread<E> {
 	@Override
 	@Nullable
 	public E poll(long time) throws InterruptedException {
+		if (!isRun()) {
+			// 停止运行时返回null, 让数据待在redis里面
+			return null;
+		}
 		// 上锁
 		lock.lockInterruptibly();
 		try {
@@ -125,6 +136,11 @@ public abstract class AbstractRedisThread<E> extends AbstractQueueThread<E> {
 					// 被唤醒. 或者超时
 					pop = get();

+					// 如果不运行了
+					if (!isRun()) {
+						break;
+					}
+
 					// 获取到值
 					if (StrUtil.isNotBlank(pop)) {
 						// 结束循环. 返回值
@@ -140,4 +156,33 @@ public abstract class AbstractRedisThread<E> extends AbstractQueueThread<E> {

 	}

+	@Override
+	public void shutdown() {
+		// 修改运行标志
+		run = false;
+		lock.lock();
+		try {
+			condition.signalAll();
+		}
+		finally {
+			lock.unlock();
+		}
+	}
+
+	@Override
+	public void shutdownHandler(List<E> list) {
+		log.warn("{} 线程被关闭! id: {}", getClass().getSimpleName(), getId());
+		for (E e : list) {
+			// 所有数据插入redis
+			put(e);
+			log.error("{}", e);
+		}
+	}
+
+	@Override
+	public boolean isRun() {
+		// 运行中 且 未被中断
+		return run && !isInterrupted();
+	}
+
 }