msb-public
/
austin
mirror of https://github.com/ZhongFuCheng3y/austin
5
3
Code Issues Projects Releases Wiki Activity

handler 线程池模块代码重构

Browse Source
pull/6/head
3y 3 years ago
parent 92cece87b1
commit 24d1bc621f
11 changed files with 68 additions and 34 deletions
Show Stats Download Patch File Download Diff File

2
austin-common/src/main/java/com/java3y/austin/common/constant/ThreadPoolConstant.java
Unescape View File

@ -25,7 +25,7 @@ public class ThreadPoolConstant {
public static final Integer COMMON_CORE_POOL_SIZE = 2;
public static final Integer COMMON_MAX_POOL_SIZE = 2;
public static final Integer COMMON_KEEP_LIVE_TIME = 60;
public static final Integer COMMON_QUEUE_SIZE = 20;
public static final Integer COMMON_QUEUE_SIZE = 128;
/**

4
austin-cron/src/main/java/com/java3y/austin/cron/utils/ReadFileUtils.java
Unescape View File

@ -1,5 +1,6 @@
package com.java3y.austin.cron.utils;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.text.csv.*;
@ -107,7 +108,8 @@ public class ReadFileUtils {
for (int j = 1; j < headerInfo.size(); j++) {
param.put(headerInfo.get(j), row.get(j));
}
result.add(CrowdInfoVo.builder().receiver(row.get(0)).params(param).build());
result.add(CrowdInfoVo.builder().receiver(CollUtil.getFirst(row.iterator())).params(param).build());
}
} catch (Exception e) {

40
austin-handler/src/main/java/com/java3y/austin/handler/config/HandlerThreadPoolConfig.java
Unescape View File

@ -0,0 +1,40 @@
package com.java3y.austin.handler.config;
import com.dtp.common.em.QueueTypeEnum;
import com.dtp.common.em.RejectedTypeEnum;
import com.dtp.core.thread.DtpExecutor;
import com.dtp.core.thread.ThreadPoolBuilder;
import com.java3y.austin.common.constant.ThreadPoolConstant;
import java.util.concurrent.TimeUnit;
/**
* handler模块 线程池的配置
*
* @author 3y
*/
public class HandlerThreadPoolConfig {
private static final String PRE_FIX = "austin.";
/**
* 业务：处理某个渠道的某种类型消息的线程池
* 配置：不丢弃消息，核心线程数不会随着keepAliveTime而减少(不会被回收)
* 动态线程池且被Spring管理：true
*
* @return
*/
public static DtpExecutor getExecutor(String groupId) {
return ThreadPoolBuilder.newBuilder()
.threadPoolName(PRE_FIX + groupId)
.corePoolSize(ThreadPoolConstant.COMMON_CORE_POOL_SIZE)
.maximumPoolSize(ThreadPoolConstant.COMMON_MAX_POOL_SIZE)
.keepAliveTime(ThreadPoolConstant.COMMON_KEEP_LIVE_TIME)
.timeUnit(TimeUnit.SECONDS)
.rejectedExecutionHandler(RejectedTypeEnum.CALLER_RUNS_POLICY.getName())
.allowCoreThreadTimeOut(false)
.workQueue(QueueTypeEnum.VARIABLE_LINKED_BLOCKING_QUEUE.getName(), ThreadPoolConstant.COMMON_QUEUE_SIZE, false)
.buildDynamic();
}
}

4
austin-handler/src/main/java/com/java3y/austin/handler/handler/EmailHandler.java → austin-handler/src/main/java/com/java3y/austin/handler/handler/impl/EmailHandler.java
Unescape View File

@ -1,4 +1,4 @@
package com.java3y.austin.handler.handler;
package com.java3y.austin.handler.handler.impl;
import cn.hutool.extra.mail.MailAccount;
@ -7,6 +7,8 @@ import com.google.common.base.Throwables;
import com.java3y.austin.common.domain.TaskInfo;
import com.java3y.austin.common.dto.EmailContentModel;
import com.java3y.austin.common.enums.ChannelType;
import com.java3y.austin.handler.handler.BaseHandler;
import com.java3y.austin.handler.handler.Handler;
import com.java3y.austin.support.utils.AccountUtils;
import com.sun.mail.util.MailSSLSocketFactory;
import lombok.extern.slf4j.Slf4j;

4
austin-handler/src/main/java/com/java3y/austin/handler/handler/OfficialAccountHandler.java → austin-handler/src/main/java/com/java3y/austin/handler/handler/impl/OfficialAccountHandler.java
Unescape View File

@ -1,10 +1,12 @@
package com.java3y.austin.handler.handler;
package com.java3y.austin.handler.handler.impl;
import com.alibaba.fastjson.JSON;
import com.google.common.base.Throwables;
import com.java3y.austin.common.domain.TaskInfo;
import com.java3y.austin.common.dto.OfficialAccountsContentModel;
import com.java3y.austin.common.enums.ChannelType;
import com.java3y.austin.handler.handler.BaseHandler;
import com.java3y.austin.handler.handler.Handler;
import com.java3y.austin.handler.script.OfficialAccountScript;
import lombok.extern.slf4j.Slf4j;
import me.chanjar.weixin.mp.bean.template.WxMpTemplateData;

4
austin-handler/src/main/java/com/java3y/austin/handler/handler/SmsHandler.java → austin-handler/src/main/java/com/java3y/austin/handler/handler/impl/SmsHandler.java
Unescape View File

@ -1,4 +1,4 @@
package com.java3y.austin.handler.handler;
package com.java3y.austin.handler.handler.impl;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
@ -8,6 +8,8 @@ import com.java3y.austin.common.domain.TaskInfo;
import com.java3y.austin.common.dto.SmsContentModel;
import com.java3y.austin.common.enums.ChannelType;
import com.java3y.austin.handler.domain.SmsParam;
import com.java3y.austin.handler.handler.BaseHandler;
import com.java3y.austin.handler.handler.Handler;
import com.java3y.austin.handler.script.SmsScript;
import com.java3y.austin.support.dao.SmsRecordDao;
import com.java3y.austin.support.domain.SmsRecord;

32
austin-handler/src/main/java/com/java3y/austin/handler/pending/TaskPendingHolder.java
Unescape View File

@ -1,12 +1,9 @@
package com.java3y.austin.handler.pending;
import com.dtp.common.em.QueueTypeEnum;
import com.dtp.common.em.RejectedTypeEnum;
import com.dtp.core.DtpRegistry;
import com.dtp.core.thread.DtpExecutor;
import com.dtp.core.thread.ThreadPoolBuilder;
import com.java3y.austin.handler.config.HandlerThreadPoolConfig;
import com.java3y.austin.handler.utils.GroupIdMappingUtils;
import com.java3y.austin.support.config.ThreadPoolExecutorShutdownDefinition;
import com.java3y.austin.support.utils.ThreadPoolUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@ -24,17 +21,9 @@ import java.util.concurrent.ExecutorService;
*/
@Component
public class TaskPendingHolder {
@Autowired
private ThreadPoolExecutorShutdownDefinition threadPoolExecutorShutdownDefinition;
private ThreadPoolUtils threadPoolUtils;
/**
* 线程池的参数(初始化参数)
*/
private Integer coreSize = 2;
private Integer maxSize = 2;
private Integer queueSize = 100;
private Map<String, ExecutorService> taskPendingHolder = new HashMap<>(32);
/**
@ -48,22 +37,15 @@ public class TaskPendingHolder {
@PostConstruct
public void init() {
/**
* example ThreadPoolName:austin-im.notice
* example ThreadPoolName:austin.im.notice
*
* 可以通过apollo配置：dynamic-tp-apollo-dtp.yml 动态修改线程池的信息
*/
for (String groupId : groupIds) {
DtpExecutor dtpExecutor = ThreadPoolBuilder.newBuilder()
.threadPoolName("austin-" + groupId)
.corePoolSize(coreSize)
.maximumPoolSize(maxSize)
.workQueue(QueueTypeEnum.LINKED_BLOCKING_QUEUE.getName(), queueSize, false)
.rejectedExecutionHandler(RejectedTypeEnum.CALLER_RUNS_POLICY.getName())
.buildDynamic();
DtpExecutor executor = HandlerThreadPoolConfig.getExecutor(groupId);
threadPoolUtils.register(executor);
DtpRegistry.register(dtpExecutor, "beanPostProcessor");
threadPoolExecutorShutdownDefinition.registryExecutor(dtpExecutor);
taskPendingHolder.put(groupId, dtpExecutor);
taskPendingHolder.put(groupId, executor);
}
}

4
austin-handler/src/main/java/com/java3y/austin/handler/receiver/Receiver.java
Unescape View File

@ -1,5 +1,6 @@
package com.java3y.austin.handler.receiver;
import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.java3y.austin.common.domain.AnchorInfo;
import com.java3y.austin.common.domain.LogParam;
@ -47,7 +48,8 @@ public class Receiver {
if (kafkaMessage.isPresent()) {
List<TaskInfo> taskInfoLists = JSON.parseArray(kafkaMessage.get(), TaskInfo.class);
String messageGroupId = GroupIdMappingUtils.getGroupIdByTaskInfo(taskInfoLists.get(0));
String messageGroupId = GroupIdMappingUtils.getGroupIdByTaskInfo(CollUtil.getFirst(taskInfoLists.iterator()));
/**
* 每个消费者组 只消费 他们自身关心的消息

3
austin-handler/src/main/java/com/java3y/austin/handler/script/TencentSmsScript.java → austin-handler/src/main/java/com/java3y/austin/handler/script/impl/TencentSmsScript.java
Unescape View File

@ -1,4 +1,4 @@
package com.java3y.austin.handler.script;
package com.java3y.austin.handler.script.impl;
import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateUtil;
@ -7,6 +7,7 @@ import cn.hutool.core.util.IdUtil;
import com.java3y.austin.common.enums.SmsStatus;
import com.java3y.austin.handler.domain.SmsParam;
import com.java3y.austin.handler.domain.TencentSmsParam;
import com.java3y.austin.handler.script.SmsScript;
import com.java3y.austin.support.domain.SmsRecord;
import com.java3y.austin.support.utils.AccountUtils;
import com.tencentcloudapi.common.Credential;

3
austin-handler/src/main/java/com/java3y/austin/handler/script/WxMpTemplateScript.java → austin-handler/src/main/java/com/java3y/austin/handler/script/impl/WxMpTemplateScript.java
Unescape View File

@ -1,5 +1,6 @@
package com.java3y.austin.handler.script;
package com.java3y.austin.handler.script.impl;
import com.java3y.austin.handler.script.OfficialAccountScript;
import lombok.extern.slf4j.Slf4j;
import me.chanjar.weixin.mp.api.WxMpService;
import me.chanjar.weixin.mp.api.impl.WxMpServiceImpl;

2
austin-web/src/main/resources/dynamic-tp-apollo-dtp.yml
Unescape View File

@ -19,7 +19,7 @@ spring:
secret: SECb544445a6a34f0315d08b17de41
receivers: 18888888888
executors:
- threadPoolName: austin-im.notice
- threadPoolName: austin.im.notice
corePoolSize: 6
maximumPoolSize: 8
queueCapacity: 200

Write Preview
Loading…
Cancel
Save