PowerJob Alarmable工作流程源碼剖析
序
本文主要研究一下PowerJob的Alarmable
Alarmable
tech/powerjob/server/extension/Alarmable.java
public interface Alarmable {
void onFailed(Alarm alarm, List<UserInfoDO> targetUserList);
}Alarmable接口定義了onFailed方法,其入?yún)閍larm及targetUserList
Alarm
public interface Alarm extends PowerSerializable {
String fetchTitle();
default String fetchContent() {
StringBuilder sb = new StringBuilder();
JSONObject content = JSONObject.parseObject(JSONObject.toJSONString(this));
content.forEach((key, originWord) -> {
sb.append(key).append(": ");
String word = String.valueOf(originWord);
if (StringUtils.endsWithIgnoreCase(key, "time") || StringUtils.endsWithIgnoreCase(key, "date")) {
try {
if (originWord instanceof Long) {
word = CommonUtils.formatTime((Long) originWord);
}
}catch (Exception ignore) {
}
}
sb.append(word).append(OmsConstant.LINE_SEPARATOR);
});
return sb.toString();
}
}Alarm定義了fetchTitle方法,提供了fetchContent默認方法,它有兩個實現(xiàn)類分別是JobInstanceAlarm、WorkflowInstanceAlarm
DingTalkAlarmService
tech/powerjob/server/extension/defaultimpl/alarm/impl/DingTalkAlarmService.java
@Slf4j
@Service
@RequiredArgsConstructor
public class DingTalkAlarmService implements Alarmable {
private final Environment environment;
private Long agentId;
private DingTalkUtils dingTalkUtils;
private Cache<String, String> mobile2UserIdCache;
private static final int CACHE_SIZE = 8192;
/**
* 防止緩存擊穿
*/
private static final String EMPTY_TAG = "EMPTY";
@Override
public void onFailed(Alarm alarm, List<UserInfoDO> targetUserList) {
if (dingTalkUtils == null) {
return;
}
Set<String> userIds = Sets.newHashSet();
targetUserList.forEach(user -> {
String phone = user.getPhone();
if (StringUtils.isEmpty(phone)) {
return;
}
try {
String userId = mobile2UserIdCache.get(phone, () -> {
try {
return dingTalkUtils.fetchUserIdByMobile(phone);
} catch (PowerJobException ignore) {
return EMPTY_TAG;
} catch (Exception ignore) {
return null;
}
});
if (!EMPTY_TAG.equals(userId)) {
userIds .add(userId);
}
}catch (Exception ignore) {
}
});
userIds.remove(null);
if (!userIds.isEmpty()) {
String userListStr = SJ.COMMA_JOINER.skipNulls().join(userIds);
List<DingTalkUtils.MarkdownEntity> markdownEntities = Lists.newLinkedList();
markdownEntities.add(new DingTalkUtils.MarkdownEntity("server", NetUtils.getLocalHost()));
String content = alarm.fetchContent().replaceAll(OmsConstant.LINE_SEPARATOR, OmsConstant.COMMA);
markdownEntities.add(new DingTalkUtils.MarkdownEntity("content", content));
try {
dingTalkUtils.sendMarkdownAsync(alarm.fetchTitle(), markdownEntities, userListStr, agentId);
}catch (Exception e) {
log.error("[DingTalkAlarmService] send ding message failed, reason is {}", e.getMessage());
}
}
}
@PostConstruct
public void init() {
String agentId = environment.getProperty(PowerJobServerConfigKey.DING_AGENT_ID);
String appKey = environment.getProperty(PowerJobServerConfigKey.DING_APP_KEY);
String appSecret = environment.getProperty(PowerJobServerConfigKey.DING_APP_SECRET);
log.info("[DingTalkAlarmService] init with appKey:{},appSecret:{},agentId:{}", appKey, appSecret, agentId);
if (StringUtils.isAnyBlank(agentId, appKey, appSecret)) {
log.warn("[DingTalkAlarmService] cannot get agentId, appKey, appSecret at the same time, this service is unavailable");
return;
}
if (!StringUtils.isNumeric(agentId)) {
log.warn("[DingTalkAlarmService] DingTalkAlarmService is unavailable due to invalid agentId: {}", agentId);
return;
}
this.agentId = Long.valueOf(agentId);
dingTalkUtils = new DingTalkUtils(appKey, appSecret);
mobile2UserIdCache = CacheBuilder.newBuilder().maximumSize(CACHE_SIZE).softValues().build();
log.info("[DingTalkAlarmService] init DingTalkAlarmService successfully!");
}
}DingTalkAlarmService實現(xiàn)了Alarmable接口,其onFailed遍歷targetUserList獲取userId,最后通過dingTalkUtils.sendMarkdownAsync發(fā)送
MailAlarmService
tech/powerjob/server/extension/defaultimpl/alarm/impl/MailAlarmService.java
@Slf4j
@Service
public class MailAlarmService implements Alarmable {
@Resource
private Environment environment;
private JavaMailSender javaMailSender;
@Value("${spring.mail.username:''}")
private String from;
@Override
public void onFailed(Alarm alarm, List<UserInfoDO> targetUserList) {
if (CollectionUtils.isEmpty(targetUserList) || javaMailSender == null || StringUtils.isEmpty(from)) {
return;
}
SimpleMailMessage sm = new SimpleMailMessage();
try {
sm.setFrom(from);
sm.setTo(targetUserList.stream().map(UserInfoDO::getEmail).filter(Objects::nonNull).toArray(String[]::new));
sm.setSubject(alarm.fetchTitle());
sm.setText(alarm.fetchContent());
javaMailSender.send(sm);
}catch (Exception e) {
log.warn("[MailAlarmService] send mail failed, reason is {}", e.getMessage());
}
}
@Autowired(required = false)
public void setJavaMailSender(JavaMailSender javaMailSender) {
this.javaMailSender = javaMailSender;
}
}MailAlarmService實現(xiàn)了Alarmable接口,其onFailed方法構建SimpleMailMessage,然后通過spring的javaMailSender.send發(fā)送
WebHookAlarmService
tech/powerjob/server/extension/defaultimpl/alarm/impl/WebHookAlarmService.java
@Slf4j
@Service
public class WebHookAlarmService implements Alarmable {
private static final String HTTP_PROTOCOL_PREFIX = "http://";
private static final String HTTPS_PROTOCOL_PREFIX = "https://";
@Override
public void onFailed(Alarm alarm, List<UserInfoDO> targetUserList) {
if (CollectionUtils.isEmpty(targetUserList)) {
return;
}
targetUserList.forEach(user -> {
String webHook = user.getWebHook();
if (StringUtils.isEmpty(webHook)) {
return;
}
// 自動添加協(xié)議頭
if (!webHook.startsWith(HTTP_PROTOCOL_PREFIX) && !webHook.startsWith(HTTPS_PROTOCOL_PREFIX)) {
webHook = HTTP_PROTOCOL_PREFIX + webHook;
}
MediaType jsonType = MediaType.parse(OmsConstant.JSON_MEDIA_TYPE);
RequestBody requestBody = RequestBody.create(jsonType, JSONObject.toJSONString(alarm));
try {
String response = HttpUtils.post(webHook, requestBody);
log.info("[WebHookAlarmService] invoke webhook[url={}] successfully, response is {}", webHook, response);
}catch (Exception e) {
log.warn("[WebHookAlarmService] invoke webhook[url={}] failed!", webHook, e);
}
});
}
}WebHookAlarmService實現(xiàn)了Alarmable接口,其onFailed方法遍歷targetUserList,挨個執(zhí)行HttpUtils.post(webHook, requestBody),用的是okhttp3來實現(xiàn)http請求回調(diào)
小結
PowerJob的Alarmable接口定義了onFailed方法,其入?yún)閍larm及targetUserList;它有三個實現(xiàn)類,分別是DingTalkAlarmService(用的是DingTalkClient)、MailAlarmService(用的是spring的JavaMailSender)、WebHookAlarmService(用的是okhttp3的OkHttpClient)。
以上就是PowerJob Alarmable工作流程源碼剖析的詳細內(nèi)容,更多關于PowerJob Alarmable的資料請關注腳本之家其它相關文章!
相關文章
Java Volatile關鍵字實現(xiàn)原理過程解析
這篇文章主要介紹了Java Volatile關鍵字實現(xiàn)原理過程解析,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-03-03
基于Java實現(xiàn)根據(jù)周次生成數(shù)據(jù)的工具類
這篇文章主要為大家詳細介紹了一個基于Java的周次生成工具,用于根據(jù)指定的年份和周次類型(自然周或財務周)生成連續(xù)的周次列表數(shù)據(jù),希望對大家有所幫助2025-08-08
Springboot如何使用OSHI獲取和操作系統(tǒng)和硬件信息
這篇文章主要介紹了Springboot如何使用OSHI獲取和操作系統(tǒng)和硬件信息問題,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-10-10
java獲取登錄者IP和登錄時間的兩種實現(xiàn)代碼詳解
這篇文章主要介紹了java獲取登錄者IP和登錄時間的實現(xiàn)代碼,本文通過兩種結合實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友可以參考下2020-07-07
SpringBoot中的@Configuration、@MapperScan注解
SpringBoot中的@Configuration和@MapperScan注解分別用于配置類和Mapper接口的自動掃描,感興趣的朋友跟隨小編一起看看吧2024-11-11
SpringBoot整合Redis使用RedisTemplate和StringRedisTemplate
Spring?Boot?Data(數(shù)據(jù))?Redis?中提供了RedisTemplate和StringRedisTemplate,其中StringRedisTemplate是RedisTemplate的子類,兩個方法基本一致。本文介紹了SpringBoot整合Redis使用RedisTemplate和StringRedisTemplate的方法,需要的可以參考一下2022-12-12

