diff --git a/README.md b/README.md
index c7d34d5..1dc8323 100644
--- a/README.md
+++ b/README.md
@@ -54,7 +54,7 @@ austin项目**核心流程**:`austin-api`接收到发送消息请求,直接
**1**、austin使用的MySQL版本**5.7x**。如果目前使用的MySQL版本8.0,注意改变`pom.xml`所依赖的版本
-**2**、适配`application.properties`的配置信息(`srping.datasource`)
+**2**、适配`application.properties`的配置信息(`spring.datasource`)
**3**、执行`sql`文件夹下的`austin.sql`创建对应的表
@@ -97,9 +97,9 @@ curl -XPOST "127.0.0.1:8080/send" -H 'Content-Type: application/json' -d '{"co
- [ ] 04、持续提高消息推送系统的影响力,让更多的业务方了解其功能,进而挖掘更多拉新和唤醒用户的玩法,提高站内的次留率和转化率
-**近期更新时间**:2022年1月25日
+**近期更新时间**:2022年2月14日
-**近期更新功能**:austin前端管理系统
+**近期更新功能**:接入xxl-job分布式定时任务框架并完成定时任务逻辑
## 项目交流
diff --git a/austin-cron/pom.xml b/austin-cron/pom.xml
index a45cc07..a784774 100644
--- a/austin-cron/pom.xml
+++ b/austin-cron/pom.xml
@@ -21,6 +21,16 @@
austin-support
0.0.1-SNAPSHOT
+
+ com.java3y.austin
+ austin-service-api
+ 0.0.1-SNAPSHOT
+
+
+ com.java3y.austin
+ austin-service-api-impl
+ 0.0.1-SNAPSHOT
+
com.xuxueli
xxl-job-core
diff --git a/austin-cron/src/main/java/com/java3y/austin/cron/constants/PendingConstant.java b/austin-cron/src/main/java/com/java3y/austin/cron/constants/PendingConstant.java
index 488d5ef..e652567 100644
--- a/austin-cron/src/main/java/com/java3y/austin/cron/constants/PendingConstant.java
+++ b/austin-cron/src/main/java/com/java3y/austin/cron/constants/PendingConstant.java
@@ -1,9 +1,12 @@
package com.java3y.austin.cron.constants;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
/**
* @author 3y
* @date 2022/2/13
- * 缓冲pending 常量
+ * 延迟缓冲 pending 常量信息
*/
public class PendingConstant {
@@ -23,9 +26,10 @@ public class PendingConstant {
public static final Long TIME_THRESHOLD = 1000L;
/**
- * 消费线程数
+ * 真正消费线程池配置的信息
*/
- public static final Integer THREAD_NUM = 2;
-
+ public static final Integer CORE_POOL_SIZE = 2;
+ public static final Integer MAX_POOL_SIZE = 2;
+ public static final BlockingQueue BLOCKING_QUEUE = new LinkedBlockingQueue<>(5);
}
diff --git a/austin-cron/src/main/java/com/java3y/austin/cron/handler/CronTaskHandler.java b/austin-cron/src/main/java/com/java3y/austin/cron/handler/CronTaskHandler.java
index 317bd20..e0b11b3 100644
--- a/austin-cron/src/main/java/com/java3y/austin/cron/handler/CronTaskHandler.java
+++ b/austin-cron/src/main/java/com/java3y/austin/cron/handler/CronTaskHandler.java
@@ -27,7 +27,6 @@ public class CronTaskHandler {
log.info("CronTaskHandler#execute messageTemplateId:{} cron exec!", XxlJobHelper.getJobParam());
Long messageTemplateId = Long.valueOf(XxlJobHelper.getJobParam());
taskHandler.handle(messageTemplateId);
-
}
}
diff --git a/austin-cron/src/main/java/com/java3y/austin/cron/pending/CrowdBatchTaskPending.java b/austin-cron/src/main/java/com/java3y/austin/cron/pending/CrowdBatchTaskPending.java
index eb24d97..8186085 100644
--- a/austin-cron/src/main/java/com/java3y/austin/cron/pending/CrowdBatchTaskPending.java
+++ b/austin-cron/src/main/java/com/java3y/austin/cron/pending/CrowdBatchTaskPending.java
@@ -1,39 +1,88 @@
package com.java3y.austin.cron.pending;
-import com.java3y.austin.cron.domain.CrowdInfoVo;
-import com.java3y.austin.support.pending.BatchPendingThread;
-import com.java3y.austin.support.pending.Pending;
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.map.MapUtil;
+import cn.hutool.core.thread.ExecutorBuilder;
+import cn.hutool.core.util.StrUtil;
+import com.google.common.collect.Lists;
+import com.java3y.austin.cron.constants.PendingConstant;
+import com.java3y.austin.cron.vo.CrowdInfoVo;
+import com.java3y.austin.service.api.domain.BatchSendRequest;
+import com.java3y.austin.service.api.domain.MessageParam;
+import com.java3y.austin.service.api.enums.BusinessCode;
+import com.java3y.austin.service.api.service.SendService;
+import com.java3y.austin.support.pending.AbstractLazyPending;
import com.java3y.austin.support.pending.PendingParam;
import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
/**
- * 批量处理任务信息
+ * 延迟批量处理人群信息
+ * 调用 batch 发送接口 进行消息推送
*
* @author 3y
*/
-@Component
@Slf4j
-public class CrowdBatchTaskPending extends Pending {
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class CrowdBatchTaskPending extends AbstractLazyPending {
- @Override
- public void initAndStart(PendingParam pendingParam) {
- threadNum = pendingParam.getThreadNum() == null ? threadNum : pendingParam.getThreadNum();
- queue = pendingParam.getQueue();
-
- for (int i = 0; i < threadNum; ++i) {
- BatchPendingThread batchPendingThread = new BatchPendingThread();
- batchPendingThread.setPendingParam(pendingParam);
- batchPendingThread.setName("batchPendingThread-" + i);
- batchPendingThread.start();
- }
+ @Autowired
+ private SendService sendService;
+
+ public CrowdBatchTaskPending() {
+ PendingParam pendingParam = new PendingParam<>();
+ pendingParam.setNumThreshold(PendingConstant.NUM_THRESHOLD)
+ .setQueue(new LinkedBlockingQueue(PendingConstant.QUEUE_SIZE))
+ .setTimeThreshold(PendingConstant.TIME_THRESHOLD)
+ .setExecutorService(ExecutorBuilder.create()
+ .setCorePoolSize(PendingConstant.CORE_POOL_SIZE)
+ .setMaxPoolSize(PendingConstant.MAX_POOL_SIZE)
+ .setWorkQueue(PendingConstant.BLOCKING_QUEUE)
+ .setHandler(new ThreadPoolExecutor.CallerRunsPolicy())
+ .build());
+ this.pendingParam = pendingParam;
}
@Override
- public void doHandle(List list) {
- log.info("theadName:{},doHandle:{}", Thread.currentThread().getName(), list.size());
+ public void doHandle(List crowdInfoVos) {
+
+ // 1. 如果参数相同,组装成同一个MessageParam发送
+ Map