xiaoguangbin/AnalysisSystemForRadionuclide
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

系统监控预警

Browse Source
This commit is contained in:
nieziyan 2023-07-10 16:42:14 +08:00
parent e8d369c46c
commit fd1dbf84f7
51 changed files with 1516 additions and 258 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
jeecg-boot-base-core/src/main/java/org/jeecg/common/api/dto/message/MessageDTO.java
View File

@ -17,7 +17,7 @@ public class MessageDTO implements Serializable {
/**
* 发送人(用户登录账户)
*/
protected String fromUser;
protected String fromUser = "admin";
/**
* 发送给(用户登录账户)

36
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/CommonConstant.java
View File

@ -490,4 +490,40 @@ public interface CommonConstant {
* 是叶子节点
*/
Integer IS_LEAF = 1;
/**
* Redis Stream Key
*/
String STREAM_KEY_WARN = "Warn:Stream_Key_Warn";
/*
* Redis Stream 消费组名
*/
String GROUP_WARN_A = "Group_Warn_A";
String GROUP_WARN_B = "Group_Warn_B";
/*
* Redis Stream 消费者名
*/
String CONSUMERWARNA1 = "Consumer_Warn_A1";
String CONSUMERWARNA2 = "Consumer_Warn_A2";
String CONSUMERWARNB1 = "Consumer_Warn_B1";
/**
* 预警规则Key前缀
*/
String PREFIX_RULE = "Warn:Rule:";
// 启用
Integer ENABLED = 1;
// 禁用
Integer DISENABLED = 0;
// 用于生成临时token的用户名
String TEMP_TOKEN_USERNAME = "admin";
// 用于生成临时token的密码
String TEMP_TOKEN_SECRET = "cb362cfeefbf3d8d";
}

6
jeecg-boot-base-core/src/main/java/org/jeecg/common/constant/enums/MessageTypeEnum.java
View File

@ -20,7 +20,11 @@ public enum MessageTypeEnum {
/** 钉钉消息 */
DD("dingtalk", "钉钉消息"),
/** 企业微信 */
QYWX("wechat_enterprise", "企业微信");
QYWX("wechat_enterprise", "企业微信"),
/* 自定义邮箱配置 */
YX("email-self", "自定义邮件消息"),
/* 手机短信 */
SMS("sms", "手机短信");
MessageTypeEnum(String type, String note){
this.type = type;

93
jeecg-boot-base-core/src/main/java/org/jeecg/common/email/EmailServiceManager.java
View File

@ -1,25 +1,34 @@
package org.jeecg.common.email;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Lists;
import com.sun.mail.imap.IMAPStore;
import com.sun.mail.smtp.SMTPAddressFailedException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ArrayUtils;
import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.email.emuns.MailContentType;
import org.jeecg.modules.base.entity.SysEmail;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Component;
import javax.mail.*;
import javax.mail.internet.AddressException;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;
import javax.mail.internet.MimeUtility;
import java.io.*;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.*;
import java.util.stream.Collectors;
/**
* 邮件服务管理器
*/
@Slf4j
@Component
public class EmailServiceManager {
private SysEmail email;
@ -119,6 +128,82 @@ public class EmailServiceManager {
return null;
}
/**
* 发送邮件
*/
public void sendMail(MessageDTO messageDTO){
// 邮箱连接属性
Properties props = new Properties();
props.put("mail.transport.protocol", "smtp");
props.put("mail.smtp.host", email.getEmailServerAddress());
props.put("mail.smtp.port", email.getPort());
props.put("mail.smtp.auth", "true");
/*props.put("mail.smtp.socketFactory.port", email.getPort());
props.put("mail.smtp.socketFactory.class", "javax.net.ssl.SSLSocketFactory");*/
Session session = Session.getInstance(props);
Transport transport = null;
Message message = new MimeMessage(session);
try {
// 邮件发送方(邮箱必须真实有效)
message.setFrom(new InternetAddress(email.getUsername()));
// 设置主题
message.setSubject(messageDTO.getTitle());
// 设置正文
message.setText(messageDTO.getContent());
// 设置收件地址
message.setRecipients(Message.RecipientType.TO,
InternetAddress.parse(messageDTO.getToUser()));
// 发送邮件
transport = session.getTransport();
transport.connect(email.getUsername(),email.getPassword());
transport.sendMessage(message,message.getAllRecipients());
} catch (MessagingException e) {
// 无效的电子邮箱导致群发失败,剔除无效邮箱然后重新发送
Address[] invalid;
if (e instanceof SMTPAddressFailedException) {
invalid = ((SMTPAddressFailedException) e).getInvalidAddresses();
reSendMail(invalid,messageDTO);
} else if (e instanceof SendFailedException) {
invalid = ((SendFailedException) e).getInvalidAddresses();
reSendMail(invalid,messageDTO);
}
}finally {
if (transport != null){
try {
transport.close();
} catch (MessagingException e) {
log.error("Transport关闭失败,{}",e);
}
}
}
}
/**
* 重新发送邮件
*
* @param invalid 无效电子邮箱列表
* @param messageDTO 消息dto
*/
private void reSendMail(Address[] invalid,MessageDTO messageDTO){
List<String> invalidList = Arrays.stream(invalid)
.map(address -> address.toString())
.collect(Collectors.toList());
log.warn("部分或者全部邮件发送失败,无效的电子邮箱:{}",invalidList);
String[] allEmails = messageDTO.getToUser().split(",");
String[] emails = new String[]{};
for (String address : invalidList) {
emails = ArrayUtil.removeEle(allEmails,address);
}
// 如果移除无效电子邮箱后,待发送邮箱不为空
String toUser = Arrays.stream(emails)
.collect(Collectors.joining(","));
if (StrUtil.isNotBlank(toUser)){
messageDTO.setToUser(toUser);
sendMail(messageDTO);
}
}
/**
* 获取邮件内容
* @param part

19
jeecg-boot-base-core/src/main/java/org/jeecg/common/exception/StreamErrorHandler.java Normal file
View File

@ -0,0 +1,19 @@
package org.jeecg.common.exception;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ErrorHandler;
/**
* 异常处理
*
* @author nieziyan
* @date 2023-06-27
*/
@Slf4j
public class StreamErrorHandler implements ErrorHandler {
@Override
public void handleError(Throwable t) {
log.error("Redis Stream消费者消费异常,该消费者消费资格已经被取消!",t);
}
}

196
jeecg-boot-base-core/src/main/java/org/jeecg/common/util/RedisStreamUtil.java Normal file
View File

@ -0,0 +1,196 @@
package org.jeecg.common.util;
import cn.hutool.core.collection.ListUtil;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.google.common.collect.Multimap;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.modules.base.dto.WarnDto;
import org.jeecg.modules.base.entity.AlarmRule;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.stereotype.Component;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
@Component
public class RedisStreamUtil{
@Autowired
private RedisTemplate<String,Object> redisTemplate;
@Autowired
private StringRedisTemplate stringRedisTemplate;
/**
* 是否有Key(判断是否有Stream)
*
* @param key
*/
public Boolean hasKey(String key){
return redisTemplate.hasKey(key);
}
/**
* 根据key获取数据
*
* @param key
*/
public Object get(String key){
return redisTemplate.opsForValue().get(key);
}
/**
* 根据streamKey获取所有消费组
*
* @param streamKey
*/
public StreamInfo.XInfoGroups getGroups(String streamKey){
return redisTemplate.opsForStream().groups(streamKey);
}
/**
* 创建消费组
*
* @param streamKey Stream Key
* @param groupName 消费组名
*/
public String createGroup(String streamKey, String groupName){
return redisTemplate.opsForStream().createGroup(streamKey, groupName);
}
/**
* 获取消费组下的所有消费者
*
* @param streamKey
* @param groupName
*/
public StreamInfo.XInfoConsumers getConsumers(String streamKey, String groupName){
return redisTemplate.opsForStream().consumers(streamKey, groupName);
}
/**
* 获取消费者的pengdings信息(没有消息内容)
*
* @param streamKey
* @param groupName
* @param consumerName
*/
public PendingMessages getPendings(String streamKey, String groupName,String consumerName){
return redisTemplate.opsForStream().pending(streamKey, Consumer.from(groupName, consumerName));
}
/**
* 读取消费者的pengdings消息(消息内容)
*
* @param streamKey
* @param groupName
* @param consumerName
*/
public List<ObjectRecord<String,WarnDto>> read(String streamKey,String groupName,String consumerName){
return stringRedisTemplate.opsForStream().read(WarnDto.class,
Consumer.from(groupName, consumerName),
StreamOffset.fromStart(streamKey));
}
/**
* 确认消费
*
* @param streamKey
* @param groupName
* @param recordIds
*/
public Long ack(String streamKey, String groupName, String... recordIds){
return redisTemplate.opsForStream().acknowledge(streamKey, groupName, recordIds);
}
/**
* 确认消费
*
* @param groupName
* @param record
*/
public Long ack(String groupName, Record<String, Object> record){
return redisTemplate.opsForStream().acknowledge(groupName, record);
}
/**
* 根据记录id删除n个消息记录
*
* @param streamKey
* @param recordIds
*/
public Long del(String streamKey, String... recordIds){
return redisTemplate.opsForStream().delete(streamKey, recordIds);
}
/**
* 添加Map消息
*
* @param streamKey
* @param value
*/
private String putMap(String streamKey, Map<String, String> value){
return redisTemplate.opsForStream().add(streamKey, value).getValue();
}
/**
* 添加Record消息
*
* @param record
*/
private String putRecord(Record<String, WarnDto> record){
/**
* 1.使用stringRedisTemplate
* 2.序列化方式使用string
**/
return stringRedisTemplate.opsForStream().add(record).getValue();
}
/**
* 向消息队列中添加Warn信息
*
* @param warnDto
*/
public String pushWarn(WarnDto warnDto){
String warnKey = CommonConstant.STREAM_KEY_WARN;
ObjectRecord<String, WarnDto> record = StreamRecords.newRecord()
.in(warnKey).ofObject(warnDto);
// 向Redis Stream中推送消息
return putRecord(record);
}
public void putRules(Multimap<String, AlarmRule> ruleMap){
Jackson2JsonRedisSerializer jacksonSerializer = new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper objectMapper = new ObjectMapper();
// 使Jackson支持Java8的新日期API
// objectMapper.registerModule(new JavaTimeModule());
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
jacksonSerializer.setObjectMapper(objectMapper);
// 返回批量命令的执行结果
List<Object> execResult = redisTemplate.executePipelined((RedisConnection connection) -> {
Set<String> keySet = ruleMap.keySet();
for (String key : keySet) {
List<AlarmRule> rules = ListUtil.toList(ruleMap.get(key));
connection.set(key.getBytes(),jacksonSerializer.serialize(rules));
}
return null;
});
}
}

16
jeecg-boot-base-core/src/main/java/org/jeecg/common/util/TokenUtils.java
View File

@ -151,4 +151,20 @@ public class TokenUtils {
}
return loginUser;
}
/**
* 获取临时令牌
*
* 模拟登陆接口，获取临时Token
*/
public static String getTempToken() {
RedisUtil redisUtil = SpringContextUtils.getBean(RedisUtil.class);
String username = CommonConstant.TEMP_TOKEN_USERNAME;
String secret = CommonConstant.TEMP_TOKEN_SECRET;
// 模拟登录生成Token
String token = JwtUtil.sign(username, secret);
// 设置Token缓存有效时间为 5 分钟
redisUtil.set(CommonConstant.PREFIX_USER_TOKEN + token, token, 5 * 60);
return token;
}
}

31
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/dto/WarnDto.java Normal file
View File

@ -0,0 +1,31 @@
package org.jeecg.modules.base.dto;
import lombok.Data;
import lombok.experimental.Accessors;
import org.jeecg.modules.base.enums.Item;
import org.jeecg.modules.base.enums.SourceType;
import java.io.Serializable;
/**
* 预警项信息
*
* @author nieziyan
* @date 2023-06-29
*/
@Data
@Accessors(chain = true)
public class WarnDto implements Serializable{
// 资源id
private String sourceId;
// 资源类型
private SourceType sourceType;
// 监控项
private Item item;
// 当前值
private double value;
}

23
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/entity/AlarmRule.java → jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/AlarmRule.java
View File

@ -1,4 +1,4 @@
package org.jeecg.modules.entity;
package org.jeecg.modules.base.entity;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
@ -10,6 +10,7 @@ import org.springframework.format.annotation.DateTimeFormat;
import java.io.Serializable;
import java.time.LocalDate;
import java.util.Date;
@Data
@TableName(value = "alarm_rule")
@ -40,7 +41,7 @@ public class AlarmRule implements Serializable {
private Integer silenceCycle;
/**
* 报警通知方式，如短信、邮件等
* 报警通知方式 1:短信 2.邮件 3.站内
*/
@TableField(value = "notification")
private String notification;
@ -69,13 +70,19 @@ public class AlarmRule implements Serializable {
@TableField(value = "source_type")
private String sourceType;
/**
* 监控项
*/
@TableField(value = "item")
private String item;
/**
* 创建时间
*/
@TableField(value = "create_time")
@JsonFormat(pattern = "yyyy-MM-dd")
@DateTimeFormat(pattern = "yyyy-MM-dd")
private LocalDate createTime;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss",timezone = "GMT+8")
@DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date createTime;
/**
* 创建人
@ -87,9 +94,9 @@ public class AlarmRule implements Serializable {
* 修改时间
*/
@TableField(value = "update_time")
@JsonFormat(pattern = "yyyy-MM-dd")
@DateTimeFormat(pattern = "yyyy-MM-dd")
private LocalDate updateTime;
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
@DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
private Date updateTime;
/**
* 修改人

13
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/Rule.java Normal file
View File

@ -0,0 +1,13 @@
package org.jeecg.modules.base.entity;
import lombok.Data;
@Data
public class Rule {
private String name;
private String operator;
private double threshold;
}

1
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/SysEmail.java
View File

@ -91,5 +91,4 @@ public class SysEmail implements Serializable {
*/
@TableField(value = "update_by")
private String updateBy;
}

21
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/enums/Item.java Normal file
View File

@ -0,0 +1,21 @@
package org.jeecg.modules.base.enums;
public enum Item {
/* 服务器相关 */
CPU_UTILIZ("cpu-utiliz"),
/* 邮箱服务相关 */
CONN("conn");
/* 数据库相关 */
private String value;
public String getValue() {
return value;
}
Item(String value) {
this.value = value;
}
}

15
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/enums/Notific.java Normal file
View File

@ -0,0 +1,15 @@
package org.jeecg.modules.base.enums;
public enum Notific {
SYSTEM("1"),EMAIL("2"),SMS("3");
private String way;
Notific(String way) {
this.way = way;
}
public String getWay() {
return way;
}
}

15
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/enums/Op.java Normal file
View File

@ -0,0 +1,15 @@
package org.jeecg.modules.base.enums;
public enum Op {
EQ("="),GT(">"),GE(">="),LT("<"),LE("<=");
private String op;
Op(String op) {
this.op = op;
}
public String getOp() {
return op;
}
}

25
jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/enums/SourceType.java Normal file
View File

@ -0,0 +1,25 @@
package org.jeecg.modules.base.enums;
/**
* 资源类型
*
* @author nieziyan
* @date 2023-06-30
*/
public enum SourceType {
EMAIL("Email"),
DATABASE("DataBase"),
SERVER("Server");
private String type;
SourceType(String type) {
this.type = type;
}
public String getType() {
return type;
}
}

24
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/controller/AlarmContactGroupMemberController.java Normal file
View File

@ -0,0 +1,24 @@
package org.jeecg.modules.controller;
import io.swagger.annotations.ApiOperation;
import org.jeecg.common.api.QueryRequest;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.entity.AlarmContactGroup;
import org.jeecg.modules.service.IAlarmContactGroupMemberService;
import org.jeecg.modules.service.IAlarmContactGroupService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
@RestController
@RequestMapping("alarmContactGroupMember")
public class AlarmContactGroupMemberController {
@Autowired
private IAlarmContactGroupMemberService alarmContactGroupMemberService;
@GetMapping("userIds")
public Result userIds(@RequestParam String groupId){
return alarmContactGroupMemberService.userIds(groupId);
}
}

3
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/controller/AlarmRuleController.java
View File

@ -4,10 +4,9 @@ import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.jeecg.common.api.QueryRequest;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.entity.AlarmRule;
import org.jeecg.modules.base.entity.AlarmRule;
import org.jeecg.modules.service.IAlarmRuleService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.*;
@RestController

4
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/controller/SysEmailController.java
View File

@ -57,4 +57,8 @@ public class SysEmailController {
return sysEmailService.findAlarmHistory(emailId, startTime, endTime);
}
@GetMapping("getSender")
public Result<SysEmail> getSender(){
return sysEmailService.getSender();
}
}

147
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/controller/SysEmailLogController.java
View File

@ -1,13 +1,13 @@
package org.jeecg.modules.controller;
import cn.hutool.core.collection.CollUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.constant.EmailConstant;
import org.jeecg.common.util.RedisStreamUtil;
import org.jeecg.modules.base.dto.WarnDto;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.modules.base.entity.SysEmailLog;
import org.jeecg.modules.base.enums.Item;
import org.jeecg.modules.base.enums.SourceType;
import org.jeecg.modules.service.ISysEmailLogService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
@ -15,13 +15,6 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
@RestController
@RequestMapping("sysEmailLog")
@Api(value = "邮箱日志", tags = "邮箱日志")
@ -32,6 +25,9 @@ public class SysEmailLogController {
@Autowired
private RedisUtil redisUtil;
@Autowired
private RedisStreamUtil redisStreamUtil;
/**
* 邮箱服务器连接状态 正常/断开
*
@ -56,35 +52,7 @@ public class SysEmailLogController {
*/
@GetMapping("total")
public Result totalEmail(@RequestParam("emailId") String emailId){
// 当日邮件量
LambdaQueryWrapper<SysEmailLog> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate today = LocalDate.now();
String todayStart = today + " 00:00:00";
String todayEnd = today + " 23:59:59";
wrapper.between(SysEmailLog::getReceiveTime,todayStart,todayEnd);
Long todayCount = sysEmailLogService.count(wrapper);
// 昨日邮件量
wrapper.clear();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate yesterday = LocalDate.now().minusDays(1);
String yesterdayStart = yesterday + " 00:00:00";
String yesterdayEnd = yesterday + " 23:59:59";
wrapper.between(SysEmailLog::getReceiveTime,yesterdayStart,yesterdayEnd);
Long yesterdayCount = sysEmailLogService.count(wrapper);
// 过去一周邮件量
wrapper.clear();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate passWeek = LocalDate.now().minusWeeks(1);
String weekStart = passWeek + " 00:00:00";
wrapper.between(SysEmailLog::getReceiveTime,weekStart,todayEnd);
Long weekCount = sysEmailLogService.count(wrapper);
Map<String,Long> result = new HashMap<>();
result.put("today",todayCount);
result.put("yesterday",yesterdayCount);
result.put("week",weekCount);
return Result.OK(result);
return sysEmailLogService.totalEmail(emailId);
}
/**
@ -94,38 +62,7 @@ public class SysEmailLogController {
*/
@GetMapping("today")
public Result today(@RequestParam("emailId") String emailId){
LambdaQueryWrapper<SysEmailLog> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate today = LocalDate.now();
String todayStart = today + " 00:00:00";
String todayEnd = today + " 23:59:59";
wrapper.between(SysEmailLog::getReceiveTime,todayStart,todayEnd);
List<SysEmailLog> emailLogs = sysEmailLogService.list(wrapper);
// 将Date转换为LocalDateTime
List<Date> allDate = emailLogs.stream()
.map(SysEmailLog::getReceiveTime)
.collect(Collectors.toList());
List<LocalDateTime> allTime = allDate.stream()
.map(item -> {
ZoneId zoneId = ZoneId.systemDefault();
return item.toInstant()
.atZone(zoneId)
.toLocalDateTime();
})
.collect(Collectors.toList());
// 按照小时分组
Map<String,Integer> statistic = new TreeMap<>();
Map<Integer, List<LocalDateTime>> groupTime = allTime.stream()
.collect(Collectors.groupingBy(LocalDateTime::getHour));
for (int i = 0; i < 24; i++) {
if(groupTime.containsKey(i)){
Integer count = groupTime.get(i).size();
statistic.put(timeStr(i),count);
}else {
statistic.put(timeStr(i),0);
}
}
return Result.OK(statistic);
return sysEmailLogService.today(emailId);
}
/**
@ -137,64 +74,16 @@ public class SysEmailLogController {
public Result analysis(@RequestParam("emailId") String emailId,
@RequestParam("startDate") String startDate,
@RequestParam("endDate") String endDate){
String startStr = startDate + " 00:00:00";
String endStr = endDate + " 23:59:59";
LambdaQueryWrapper<SysEmailLog> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(SysEmailLog::getEmailId,emailId);
wrapper.between(SysEmailLog::getReceiveTime,startStr,endStr);
Set<String> xData = new HashSet<>();
Collection<Integer> yData = new ArrayList<>();
Map<String,Integer> statistic = new TreeMap<>();
Map<String,Object> result = new HashMap<>();
List<Date> allDate = sysEmailLogService.listObjs(wrapper,
emailLog -> ((SysEmailLog) emailLog).getReceiveTime());
// 将Date转换为LocalDateTime
List<LocalDateTime> allTime = allDate.stream()
.map(item -> {
ZoneId zoneId = ZoneId.systemDefault();
return item.toInstant()
.atZone(zoneId)
.toLocalDateTime();
})
.collect(Collectors.toList());
if (CollUtil.isEmpty(allDate)){
result.put("xData",xData);
result.put("yData",yData);
return Result.OK(result);
}
/* 支持跨年跨月选择 */
// 通过年月日进行分组 例:2023-06-06
DateTimeFormatter formatter = DateTimeFormatter.ISO_LOCAL_DATE;
Map<String, List<LocalDateTime>> group = allTime.stream()
.collect(Collectors.groupingBy(datetime -> datetime.format(formatter)));
// 列举出startDate-endDate中所有日期(包括闰年)
// 没有邮件的日期,对应的值设置为0
LocalDate start = LocalDate.parse(startDate);
LocalDate end = LocalDate.parse(endDate);
while (!start.isAfter(end)) {
String key = start.format(formatter);
if (group.containsKey(key)){
Integer count = group.get(key).size();
statistic.put(key,count);
}else {
statistic.put(key,0);
}
start = start.plusDays(1);
}
// 返回结果
xData = statistic.keySet();
yData = statistic.values();
result.put("xData",xData);
result.put("yData",yData);
return Result.OK(result);
return sysEmailLogService.analysis(emailId, startDate, endDate);
}
private String timeStr(Integer time){
if (time < 10){
return "0" + time + ":00";
}
return time + ":00";
@GetMapping("test")
public void test(String groupId){
// sendMessage.send(groupId);
WarnDto warnDto = new WarnDto();
warnDto.setSourceType(SourceType.DATABASE).setValue(80d)
.setSourceId("1").setItem(Item.CPU_UTILIZ);
redisStreamUtil.pushWarn(warnDto);
}
}

12
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/IAlarmSysUserService.java → jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/feignclient/SystemClient.java
View File

@ -1,17 +1,19 @@
package org.jeecg.modules.service;
package org.jeecg.modules.feignclient;
import org.jeecg.modules.entity.SysUser;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import java.util.List;
import java.util.Map;
@Component
@FeignClient(value = "jeecg-system")
public interface IAlarmSysUserService {
@FeignClient("jeecg-system")
public interface SystemClient {
/* 系统用户相关 */
@RequestMapping("/sys/user/findUserMap")
Map<String, SysUser> findUserMap();
}

2
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/mapper/AlarmRuleMapper.java
View File

@ -1,7 +1,7 @@
package org.jeecg.modules.mapper;
import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import org.jeecg.modules.entity.AlarmRule;
import org.jeecg.modules.base.entity.AlarmRule;
public interface AlarmRuleMapper extends BaseMapper<AlarmRule> {
}

12
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/IAlarmContactGroupMemberService.java Normal file
View File

@ -0,0 +1,12 @@
package org.jeecg.modules.service;
import com.baomidou.mybatisplus.extension.service.IService;
import org.jeecg.common.api.QueryRequest;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.entity.AlarmContactGroup;
import org.jeecg.modules.entity.AlarmContactGroupMember;
import org.springframework.web.bind.annotation.RequestParam;
public interface IAlarmContactGroupMemberService extends IService<AlarmContactGroupMember> {
Result userIds(String groupId);
}

4
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/IAlarmRuleService.java
View File

@ -3,7 +3,8 @@ package org.jeecg.modules.service;
import com.baomidou.mybatisplus.extension.service.IService;
import org.jeecg.common.api.QueryRequest;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.entity.AlarmRule;
import org.jeecg.modules.base.dto.WarnDto;
import org.jeecg.modules.base.entity.AlarmRule;
public interface IAlarmRuleService extends IService<AlarmRule> {
@ -16,5 +17,4 @@ public interface IAlarmRuleService extends IService<AlarmRule> {
Result update(AlarmRule alarmRule);
Result deleteById(String id);
}

7
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/ISysEmailLogService.java
View File

@ -1,7 +1,14 @@
package org.jeecg.modules.service;
import com.baomidou.mybatisplus.extension.service.IService;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.base.entity.SysEmailLog;
import org.springframework.web.bind.annotation.RequestParam;
public interface ISysEmailLogService extends IService<SysEmailLog> {
Result totalEmail(String emailId);
Result today(String emailId);
Result analysis(String emailId, String startDate, String endDate);
}

2
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/ISysEmailService.java
View File

@ -21,4 +21,6 @@ public interface ISysEmailService extends IService<SysEmail> {
Result findAlarmHistory(String emailId, Date startTime, Date endTime);
Result<SysEmail> getSender();
}

26
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/impl/AlarmContactGroupMemberServiceImpl.java Normal file
View File

@ -0,0 +1,26 @@
package org.jeecg.modules.service.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.entity.AlarmContactGroupMember;
import org.jeecg.modules.mapper.AlarmContactGroupMemberMapper;
import org.jeecg.modules.service.IAlarmContactGroupMemberService;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.stream.Collectors;
@Service
public class AlarmContactGroupMemberServiceImpl extends ServiceImpl<AlarmContactGroupMemberMapper, AlarmContactGroupMember> implements IAlarmContactGroupMemberService {
@Override
public Result userIds(String groupId) {
LambdaQueryWrapper<AlarmContactGroupMember> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(AlarmContactGroupMember::getGroupId,groupId);
List<String> userIds = this.list(wrapper).stream()
.map(AlarmContactGroupMember::getUserId)
.collect(Collectors.toList());
return Result.OK(userIds);
}
}

7
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/impl/AlarmContactGroupServiceImpl.java
View File

@ -12,13 +12,12 @@ import org.jeecg.common.util.SpringContextUtils;
import org.jeecg.modules.entity.AlarmContactGroup;
import org.jeecg.modules.entity.AlarmContactGroupMember;
import org.jeecg.modules.entity.SysUser;
import org.jeecg.modules.feignclient.SystemClient;
import org.jeecg.modules.mapper.AlarmContactGroupMapper;
import org.jeecg.modules.mapper.AlarmContactGroupMemberMapper;
import org.jeecg.modules.service.IAlarmContactGroupService;
import org.jeecg.modules.service.IAlarmSysUserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.servlet.http.HttpServletRequest;
import java.time.LocalDate;
import java.util.*;
@ -30,13 +29,13 @@ public class AlarmContactGroupServiceImpl extends ServiceImpl<AlarmContactGroupM
@Autowired
private AlarmContactGroupMemberMapper alarmContactGroupMemberMapper;
@Autowired
private IAlarmSysUserService alarmSysUserService;
private SystemClient systemClient;
@Override
public Result findPage(QueryRequest queryRequest, AlarmContactGroup alarmContactGroup) {
Result result = new Result();
//获取用户信息
Map<String ,SysUser> userList = alarmSysUserService.findUserMap();
Map<String ,SysUser> userList = systemClient.findUserMap();
Page<AlarmContactGroup> page = new Page<>();
LambdaQueryWrapper<AlarmContactGroup> queryWrapper = new LambdaQueryWrapper<>();
Page<AlarmContactGroup> alarmContactGroupPage = this.baseMapper.selectPage(page, queryWrapper);

62
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/impl/AlarmRuleServiceImpl.java
View File

@ -1,29 +1,48 @@
package org.jeecg.modules.service.impl;
import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import io.netty.util.internal.StringUtil;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
import com.google.common.collect.Multiset;
import org.jeecg.common.api.QueryRequest;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.system.util.JwtUtil;
import org.jeecg.common.util.RedisStreamUtil;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.common.util.SpringContextUtils;
import org.jeecg.modules.entity.AlarmRule;
import org.jeecg.modules.base.dto.WarnDto;
import org.jeecg.modules.base.entity.AlarmRule;
import org.jeecg.modules.base.enums.Op;
import org.jeecg.modules.base.enums.SourceType;
import org.jeecg.modules.mapper.AlarmRuleMapper;
import org.jeecg.modules.service.IAlarmRuleService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.servlet.http.HttpServletRequest;
import java.time.LocalDate;
import java.util.Date;
import java.util.Objects;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
@Service("alarmRuleService")
public class AlarmRuleServiceImpl extends ServiceImpl<AlarmRuleMapper, AlarmRule> implements IAlarmRuleService {
@Autowired
private RedisUtil redisUtil;
@Autowired
private RedisStreamUtil redisStreamUtil;
@Override
public Result findPage(QueryRequest queryRequest, AlarmRule alarmRule) {
Result result = new Result();
@ -62,7 +81,7 @@ public class AlarmRuleServiceImpl extends ServiceImpl<AlarmRuleMapper, AlarmRule
String username = JwtUtil.getUserNameByToken(request);
Long id = IdWorker.getId();
alarmRule.setId(id.toString());
alarmRule.setCreateTime(LocalDate.now());
alarmRule.setCreateTime(new Date());
alarmRule.setCreateBy(username);
if (StringUtils.isNotBlank(alarmRule.getOperator())){
String jsonString = JSON.toJSONString(alarmRule.getOperator());
@ -71,6 +90,8 @@ public class AlarmRuleServiceImpl extends ServiceImpl<AlarmRuleMapper, AlarmRule
this.baseMapper.insert(alarmRule);
result.setSuccess(true);
result.success("新增成功");
// 同步Redis
rule2Redis();
return result;
}
@ -88,7 +109,7 @@ public class AlarmRuleServiceImpl extends ServiceImpl<AlarmRuleMapper, AlarmRule
result.error500("对应数据不存在");
return result;
}
alarmRule.setUpdateTime(LocalDate.now());
alarmRule.setUpdateTime(new Date());
alarmRule.setUpdateBy(username);
if (StringUtils.isNotBlank(alarmRule.getOperator())){
String jsonString = JSON.toJSONString(alarmRule.getOperator());
@ -97,6 +118,8 @@ public class AlarmRuleServiceImpl extends ServiceImpl<AlarmRuleMapper, AlarmRule
this.baseMapper.updateById(alarmRule);
result.setSuccess(true);
result.success("修改成功");
// 同步Redis
rule2Redis();
return result;
}
@ -106,7 +129,34 @@ public class AlarmRuleServiceImpl extends ServiceImpl<AlarmRuleMapper, AlarmRule
this.baseMapper.deleteById(id);
result.setSuccess(true);
result.success("删除成功");
// 同步Redis
rule2Redis();
return result;
}
/*
* 将所有警告规则放到Redis缓存
* 1.规则表数据变动
* 2.系统启动后
**/
@PostConstruct
public void rule2Redis(){
String prefix = CommonConstant.PREFIX_RULE;
LambdaQueryWrapper<AlarmRule> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(AlarmRule::getEnabled,1);
List<AlarmRule> alarmRules = this.list(wrapper);
Multimap<String, AlarmRule> ruleMap = ArrayListMultimap.create();
for (AlarmRule alarmRule : alarmRules) {
String sourceType = alarmRule.getSourceType();
String sourceId = alarmRule.getSourceId();
String item = alarmRule.getItem();
String key = prefix + sourceType + "_" + sourceId + "_" + item;
ruleMap.put(key,alarmRule);
}
redisStreamUtil.putRules(ruleMap);
}
public static void main(String[] args) {
}
}

142
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/impl/SysEmailLogServiceImpl.java
View File

@ -1,11 +1,153 @@
package org.jeecg.modules.service.impl;
import cn.hutool.core.collection.CollUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.base.entity.SysEmailLog;
import org.jeecg.modules.mapper.SysEmailLogMapper;
import org.jeecg.modules.service.ISysEmailLogService;
import org.springframework.stereotype.Service;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
@Service("sysEmailLogService")
public class SysEmailLogServiceImpl extends ServiceImpl<SysEmailLogMapper, SysEmailLog> implements ISysEmailLogService {
@Override
public Result totalEmail(String emailId) {
// 当日邮件量
LambdaQueryWrapper<SysEmailLog> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate today = LocalDate.now();
String todayStart = today + " 00:00:00";
String todayEnd = today + " 23:59:59";
wrapper.between(SysEmailLog::getReceiveTime,todayStart,todayEnd);
Long todayCount = this.count(wrapper);
// 昨日邮件量
wrapper.clear();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate yesterday = LocalDate.now().minusDays(1);
String yesterdayStart = yesterday + " 00:00:00";
String yesterdayEnd = yesterday + " 23:59:59";
wrapper.between(SysEmailLog::getReceiveTime,yesterdayStart,yesterdayEnd);
Long yesterdayCount = this.count(wrapper);
// 过去一周邮件量
wrapper.clear();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate passWeek = LocalDate.now().minusWeeks(1);
String weekStart = passWeek + " 00:00:00";
wrapper.between(SysEmailLog::getReceiveTime,weekStart,todayEnd);
Long weekCount = this.count(wrapper);
Map<String,Long> result = new HashMap<>();
result.put("today",todayCount);
result.put("yesterday",yesterdayCount);
result.put("week",weekCount);
return Result.OK(result);
}
@Override
public Result today(String emailId) {
LambdaQueryWrapper<SysEmailLog> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(SysEmailLog::getEmailId,emailId);
LocalDate today = LocalDate.now();
String todayStart = today + " 00:00:00";
String todayEnd = today + " 23:59:59";
wrapper.between(SysEmailLog::getReceiveTime,todayStart,todayEnd);
List<SysEmailLog> emailLogs = this.list(wrapper);
// 将Date转换为LocalDateTime
List<Date> allDate = emailLogs.stream()
.map(SysEmailLog::getReceiveTime)
.collect(Collectors.toList());
List<LocalDateTime> allTime = allDate.stream()
.map(item -> {
ZoneId zoneId = ZoneId.systemDefault();
return item.toInstant()
.atZone(zoneId)
.toLocalDateTime();
})
.collect(Collectors.toList());
// 按照小时分组
Map<String,Integer> statistic = new TreeMap<>();
Map<Integer, List<LocalDateTime>> groupTime = allTime.stream()
.collect(Collectors.groupingBy(LocalDateTime::getHour));
for (int i = 0; i < 24; i++) {
if(groupTime.containsKey(i)){
Integer count = groupTime.get(i).size();
statistic.put(timeStr(i),count);
}else {
statistic.put(timeStr(i),0);
}
}
return Result.OK(statistic);
}
@Override
public Result analysis(String emailId, String startDate, String endDate) {
String startStr = startDate + " 00:00:00";
String endStr = endDate + " 23:59:59";
LambdaQueryWrapper<SysEmailLog> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(SysEmailLog::getEmailId,emailId);
wrapper.between(SysEmailLog::getReceiveTime,startStr,endStr);
Set<String> xData = new HashSet<>();
Collection<Integer> yData = new ArrayList<>();
Map<String,Integer> statistic = new TreeMap<>();
Map<String,Object> result = new HashMap<>();
List<Date> allDate = this.listObjs(wrapper,
emailLog -> ((SysEmailLog) emailLog).getReceiveTime());
// 将Date转换为LocalDateTime
List<LocalDateTime> allTime = allDate.stream()
.map(item -> {
ZoneId zoneId = ZoneId.systemDefault();
return item.toInstant()
.atZone(zoneId)
.toLocalDateTime();
})
.collect(Collectors.toList());
if (CollUtil.isEmpty(allDate)){
result.put("xData",xData);
result.put("yData",yData);
return Result.OK(result);
}
/* 支持跨年跨月选择 */
// 通过年月日进行分组 例:2023-06-06
DateTimeFormatter formatter = DateTimeFormatter.ISO_LOCAL_DATE;
Map<String, List<LocalDateTime>> group = allTime.stream()
.collect(Collectors.groupingBy(datetime -> datetime.format(formatter)));
// 列举出startDate-endDate中所有日期(包括闰年)
// 没有邮件的日期,对应的值设置为0
LocalDate start = LocalDate.parse(startDate);
LocalDate end = LocalDate.parse(endDate);
while (!start.isAfter(end)) {
String key = start.format(formatter);
if (group.containsKey(key)){
Integer count = group.get(key).size();
statistic.put(key,count);
}else {
statistic.put(key,0);
}
start = start.plusDays(1);
}
// 返回结果
xData = statistic.keySet();
yData = statistic.values();
result.put("xData",xData);
result.put("yData",yData);
return Result.OK(result);
}
private String timeStr(Integer time){
if (time < 10){
return "0" + time + ":00";
}
return time + ":00";
}
}

15
jeecg-module-abnormal-alarm/src/main/java/org/jeecg/modules/service/impl/SysEmailServiceImpl.java
View File

@ -6,6 +6,8 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.jeecg.common.api.QueryRequest;
import org.jeecg.common.api.vo.Result;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.email.emuns.SysMailType;
import org.jeecg.common.system.util.JwtUtil;
import org.jeecg.common.util.DateUtils;
import org.jeecg.common.util.SpringContextUtils;
@ -133,4 +135,17 @@ public class SysEmailServiceImpl extends ServiceImpl<SysEmailMapper, SysEmail> i
return result;
}
/**
* 获取邮件发送服务器信息
*/
@Override
public Result<SysEmail> getSender() {
LambdaQueryWrapper<SysEmail> wrapper = new LambdaQueryWrapper<>();
wrapper.eq(SysEmail::getEmilType, SysMailType.SEND_EMAIL.getEmailType());
wrapper.eq(SysEmail::getEnabled, CommonConstant.ENABLED);
List<SysEmail> emails = this.list(wrapper);
SysEmail sysEmail = emails.stream().findFirst().get();
return Result.OK(sysEmail);
}
}

12
jeecg-module-system/jeecg-system-biz/pom.xml
View File

@ -39,6 +39,18 @@
<groupId>org.jeecgframework.jimureport</groupId>
<artifactId>jimureport-nosql-starter</artifactId>
</dependency>-->
<!-- 引入jeecg-boot-starter-cloud依赖 -->
<dependency>
<groupId>org.jeecgframework.boot</groupId>
<artifactId>jeecg-boot-starter-cloud</artifactId>
<!-- 3.2版本号后，可选择是否排除jeecg-system-cloud-api，不排除会优先通过fegin调用接口
<exclusions>
<exclusion>
<groupId>org.jeecgframework.boot</groupId>
<artifactId>jeecg-system-cloud-api</artifactId>
</exclusion>
</exclusions>-->
</dependency>
</dependencies>
</project>

1
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/api/controller/SystemApiController.java
View File

@ -30,6 +30,7 @@ public class SystemApiController {
@Autowired
private SysBaseApiImpl sysBaseApi;
@Autowired
private ISysUserService sysUserService;

23
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/feignclient/AbnormalAlarmClient.java Normal file
View File

@ -0,0 +1,23 @@
package org.jeecg.modules.feignclient;
import org.jeecg.common.api.vo.Result;
import org.jeecg.modules.base.entity.SysEmail;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import java.util.List;
@Component
@FeignClient("jeecg-abnormal-alarm")
public interface AbnormalAlarmClient {
/* SysEmailController下相关接口 */
@GetMapping("/sysEmail/getSender")
Result<SysEmail> getSender();
/* AlarmContactGroupMemberController下相关接口 */
@GetMapping("/alarmContactGroupMember/userIds")
Result<List<String>> userIds(@RequestParam String groupId);
}

122
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/message/SendMessage.java Normal file
View File

@ -0,0 +1,122 @@
package org.jeecg.modules.message;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.config.mqtoken.UserTokenContext;
import org.jeecg.common.constant.enums.MessageTypeEnum;
import org.jeecg.common.system.api.ISysBaseAPI;
import org.jeecg.modules.base.entity.Rule;
import org.jeecg.modules.base.enums.Notific;
import org.jeecg.modules.feignclient.AbnormalAlarmClient;
import org.jeecg.modules.system.entity.SysUser;
import org.jeecg.modules.system.service.ISysUserService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import static org.jeecg.common.util.TokenUtils.getTempToken;
/**
* 消息推送 (站内|邮箱|短信)
*
* @author nieziyan
* @date 2023-06-21
*/
@Component
public class SendMessage {
private final String SYSTEM = "Username";
private final String EMAIL = "Email";
private final String SMS = "Phone";
@Autowired
private ISysBaseAPI sysBaseAPI;
@Autowired
private ISysUserService sysUserService;
@Autowired
private AbnormalAlarmClient abnormalAlarmClient;
/**
* 根据联系人组id向用户推送消息
*
* @param groupId
*/
public void send(String groupId, String notific, Rule rule){
// start:生成临时Token到线程中
UserTokenContext.setToken(getTempToken());
// 封装MessageDTO消息体
MessageDTO messageDTO = new MessageDTO();
messageDTO.setTitle("系统预警消息");
messageDTO.setContent("一封测试邮件");
Map<String, String> contact = getContact(groupId);
if (StrUtil.isBlank(notific))return;
String[] ways = notific.split(",");
for (String way : ways) {
if(way.equals(Notific.SYSTEM.getWay())){// 1.推送系统消息
String toSys = contact.get(SYSTEM);
if (StrUtil.isNotBlank(toSys)){
messageDTO.setToUser(toSys);
messageDTO.setType(MessageTypeEnum.XT.getType());
sysBaseAPI.sendTemplateMessage(messageDTO);
}
} else if (way.equals(Notific.EMAIL.getWay())) {// 2.推送邮箱
String toEmail = contact.get(EMAIL);
if (StrUtil.isNotBlank(toEmail)){
messageDTO.setToUser(toEmail);
messageDTO.setType(MessageTypeEnum.YX.getType());
sysBaseAPI.sendTemplateMessage(messageDTO);
}
} else if (way.equals(Notific.SMS.getWay())) {// 3.推送短信
String toSms = contact.get(SMS);
if (StrUtil.isNotBlank(toSms)){
messageDTO.setToUser(toSms);
messageDTO.setType(MessageTypeEnum.SMS.getType());
sysBaseAPI.sendTemplateMessage(messageDTO);
}
}
}
// end:删除临时Token
UserTokenContext.remove();
}
/**
* 获取联系方式
*
* @param groupId
*/
private Map<String, String> getContact(String groupId){
List<String> userIds = abnormalAlarmClient.userIds(groupId).getResult();
// 查询用户信息
List<SysUser> sysUsers = sysUserService.listByIds(userIds);
// 用户名
String usernameList = sysUsers.stream()
.filter(user -> StrUtil.isNotBlank(user.getUsername()))
.map(SysUser::getUsername)
.collect(Collectors.joining(","));
// 邮箱
String emailList = sysUsers.stream()
.filter(user -> StrUtil.isNotBlank(user.getEmail()))
.map(SysUser::getEmail)
.collect(Collectors.joining(","));
// 手机号码
String phoneList = sysUsers.stream()
.filter(user -> StrUtil.isNotBlank(user.getPhone()))
.map(SysUser::getPhone)
.collect(Collectors.joining(","));
Map<String,String> result = new HashMap<>();
result.put(SYSTEM,usernameList);
result.put(EMAIL,emailList);
result.put(SMS,phoneList);
return result;
}
}

42
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/message/handle/impl/EmailPushMsgHandle.java Normal file
View File

@ -0,0 +1,42 @@
package org.jeecg.modules.message.handle.impl;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.email.EmailServiceManager;
import org.jeecg.modules.base.entity.SysEmail;
import org.jeecg.modules.feignclient.AbnormalAlarmClient;
import org.jeecg.modules.message.handle.ISendMsgHandle;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.mail.MessagingException;
import java.io.UnsupportedEncodingException;
import static org.jeecg.common.email.EmailServiceManager.getInstance;
/**
* @Description: 邮箱发送信息
* @author: nieziyan
*/
@Slf4j
@Component("emailPushMsgHandle")
public class EmailPushMsgHandle implements ISendMsgHandle {
@Autowired
private AbnormalAlarmClient alarmClient;
@Override
public void sendMessage(MessageDTO messageDTO) {
// 获取邮件发送服务器信息
SysEmail sysEmail = alarmClient.getSender().getResult();
// 初始化邮件服务
EmailServiceManager emailService = getInstance();
emailService.init(sysEmail);
// 发送邮件
emailService.sendMail(messageDTO);
}
@Override
public void sendMsg(String toEmail, String title, String content) {}
}

10
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/message/handle/impl/SmsSendMsgHandle.java
View File

@ -1,6 +1,8 @@
package org.jeecg.modules.message.handle.impl;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.util.DySmsHelper;
import org.jeecg.modules.message.handle.ISendMsgHandle;
/**
@ -11,9 +13,11 @@ import org.jeecg.modules.message.handle.ISendMsgHandle;
public class SmsSendMsgHandle implements ISendMsgHandle {
@Override
public void sendMsg(String esReceiver, String esTitle, String esContent) {
// TODO Auto-generated method stub
log.info("发短信");
public void sendMessage(MessageDTO messageDTO){
}
@Override
public void sendMsg(String esReceiver, String esTitle, String esContent) {}
}

35
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/message/handle/impl/SystemSendMsgHandle.java
View File

@ -1,5 +1,6 @@
package org.jeecg.modules.message.handle.impl;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import org.jeecg.common.api.dto.message.MessageDTO;
import org.jeecg.common.constant.CommonConstant;
@ -105,27 +106,23 @@ public class SystemSendMsgHandle implements ISendMsgHandle {
announcement.setDelFlag(String.valueOf(CommonConstant.DEL_FLAG_0));
sysAnnouncementMapper.insert(announcement);
// 2.插入用户通告阅读标记表记录
String userId = toUser;
String[] userIds = userId.split(",");
String anntId = announcement.getId();
for(int i=0;i<userIds.length;i++) {
if(oConvertUtils.isNotEmpty(userIds[i])) {
SysUser sysUser = userMapper.getUserByName(userIds[i]);
if(sysUser==null) {
continue;
}
SysAnnouncementSend announcementSend = new SysAnnouncementSend();
announcementSend.setAnntId(anntId);
announcementSend.setUserId(sysUser.getId());
announcementSend.setReadFlag(CommonConstant.NO_READ_FLAG);
sysAnnouncementSendMapper.insert(announcementSend);
JSONObject obj = new JSONObject();
obj.put(WebsocketConst.MSG_CMD, WebsocketConst.CMD_USER);
obj.put(WebsocketConst.MSG_USER_ID, sysUser.getId());
obj.put(WebsocketConst.MSG_ID, announcement.getId());
obj.put(WebsocketConst.MSG_TXT, announcement.getTitile());
webSocket.sendMessage(sysUser.getId(), obj.toJSONString());
if(StrUtil.isNotBlank(toUser)) {
SysUser sysUser = userMapper.getUserByName(toUser);
if(sysUser==null) {
return;
}
SysAnnouncementSend announcementSend = new SysAnnouncementSend();
announcementSend.setAnntId(anntId);
announcementSend.setUserId(sysUser.getId());
announcementSend.setReadFlag(CommonConstant.NO_READ_FLAG);
sysAnnouncementSendMapper.insert(announcementSend);
JSONObject obj = new JSONObject();
obj.put(WebsocketConst.MSG_CMD, WebsocketConst.CMD_USER);
obj.put(WebsocketConst.MSG_USER_ID, sysUser.getId());
obj.put(WebsocketConst.MSG_ID, announcement.getId());
obj.put(WebsocketConst.MSG_TXT, announcement.getTitile());
webSocket.sendMessage(sysUser.getId(), obj.toJSONString());
}
}
}

12
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/job/DatabaseJob.java Normal file
View File

@ -0,0 +1,12 @@
package org.jeecg.modules.quartz.job;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
public class DatabaseJob implements Job {
@Override
public void execute(JobExecutionContext context) throws JobExecutionException {
}
}

23
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/job/SampleJob.java
View File

@ -1,23 +0,0 @@
package org.jeecg.modules.quartz.job;
import org.jeecg.common.util.DateUtils;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import lombok.extern.slf4j.Slf4j;
/**
* 示例不带参定时任务
*
* @Author Scott
*/
@Slf4j
public class SampleJob implements Job {
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
log.info(" Job Execution key："+jobExecutionContext.getJobDetail().getKey());
log.info(String.format(" Jeecg-Boot 普通定时任务 SampleJob ! 时间:" + DateUtils.getTimestamp()));
}
}

32
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/job/SampleParamJob.java
View File

@ -1,32 +0,0 @@
package org.jeecg.modules.quartz.job;
import org.jeecg.common.util.DateUtils;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import lombok.extern.slf4j.Slf4j;
/**
* 示例带参定时任务
*
* @Author Scott
*/
@Slf4j
public class SampleParamJob implements Job {
/**
* 若参数变量名修改 QuartzJobController中也需对应修改
*/
private String parameter;
public void setParameter(String parameter) {
this.parameter = parameter;
}
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
log.info(" Job Execution key："+jobExecutionContext.getJobDetail().getKey());
log.info( String.format("welcome %s! Jeecg-Boot 带参数定时任务 SampleParamJob ! 时间:" + DateUtils.now(), this.parameter));
}
}

46
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/job/SysInfoJob.java Normal file
View File

@ -0,0 +1,46 @@
package org.jeecg.modules.quartz.job;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.util.RedisStreamUtil;
import org.jeecg.common.util.SpringContextHolder;
import org.jeecg.common.util.SpringContextUtils;
import org.jeecg.modules.base.dto.WarnDto;
import org.jeecg.modules.base.enums.Item;
import org.jeecg.modules.base.enums.SourceType;
import org.jeecgframework.core.util.ApplicationContextUtil;
import org.quartz.*;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.Map;
@Data
@Slf4j
@PersistJobDataAfterExecution
@DisallowConcurrentExecution
public class SysInfoJob implements Job {
private String parameter;
/**
* 根据host定时查询服务器信息
* 并向消息队列中推送信息
*
* @param context
*/
@Override
public void execute(JobExecutionContext context) throws JobExecutionException {
RedisStreamUtil redisStreamUtil = SpringContextUtils.getBean(RedisStreamUtil.class);
/* 获取服务器信息并推送 */
log.info("监控的主机地址:{}",parameter);
WarnDto warnDto = new WarnDto();
warnDto.setSourceId("2").setSourceType(SourceType.EMAIL)
.setItem(Item.CONN).setValue(0d);
String id = redisStreamUtil.pushWarn(warnDto);
log.info("新增RecordId:{}",id);
}
}

2
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/quartz/service/impl/QuartzJobServiceImpl.java
View File

@ -25,8 +25,10 @@ import java.util.List;
@Slf4j
@Service
public class QuartzJobServiceImpl extends ServiceImpl<QuartzJobMapper, QuartzJob> implements IQuartzJobService {
@Autowired
private QuartzJobMapper quartzJobMapper;
@Autowired
private Scheduler scheduler;

142
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/redisStream/ConsumeA1.java Normal file
View File

@ -0,0 +1,142 @@
package org.jeecg.modules.redisStream;
import cn.hutool.core.util.StrUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.Value;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.util.RedisStreamUtil;
import org.jeecg.common.util.SpringContextUtils;
import org.jeecg.modules.base.dto.WarnDto;
import org.jeecg.modules.base.entity.AlarmRule;
import org.jeecg.modules.base.entity.Rule;
import org.jeecg.modules.base.enums.Item;
import org.jeecg.modules.base.enums.Op;
import org.jeecg.modules.base.enums.SourceType;
import org.jeecg.modules.message.SendMessage;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.PendingMessage;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Iterator;
import java.util.List;
import static org.jeecg.modules.base.enums.Op.*;
@Data
@Slf4j
@Component
@NoArgsConstructor
public class ConsumeA1 implements StreamListener<String, ObjectRecord<String, WarnDto>> {
private String groupName;
private String consumerName;
private SendMessage sendMessage;
private RedisStreamUtil redisStreamUtil;
public ConsumeA1(String groupName, String consumerName) {
this.groupName = groupName;
this.consumerName = consumerName;
}
@Override
public void onMessage(ObjectRecord<String,WarnDto> message) {
/* 避免消费抛出异常后,取消此消费者的消费资格 */
try {
String streamKey = message.getStream();
init();
/**
* 新消息在未进行ACK之前,状态也为pending,
* 直接消费所有异常未确认的消息和新消息
*/
List<ObjectRecord<String, WarnDto>> pendingList = redisStreamUtil.read(streamKey, groupName, consumerName);
for (ObjectRecord<String, WarnDto> record : pendingList) {
RecordId recordId = record.getId();
WarnDto warnDto = record.getValue();
// 消费消息
consume(warnDto);
// 消费完成后,手动确认消费消息[消息消费成功]
// 否则就是消费抛出异常,进入pending_ids[消息消费失败]
redisStreamUtil.ack(streamKey, groupName, recordId.getValue());
// 手动删除已消费消息
redisStreamUtil.del(streamKey, recordId.getValue());
}
}catch (JsonProcessingException e) {
log.error("消费者[{}]消费异常,报警规则解析失败!",consumerName);
e.printStackTrace();
}catch (RuntimeException e){
log.error("消费者[{}]消费异常,请及时排查原因!",consumerName);
e.printStackTrace();
}
}
/**
* 消费方法,根据校验情况发送预警信息
*
* @param warnDto
*/
private void consume(WarnDto warnDto) throws JsonProcessingException {
String sourceType = warnDto.getSourceType() == null
? "" : warnDto.getSourceType().getType();
String sourceId = warnDto.getSourceId();
String item = warnDto.getItem() == null
? "" : warnDto.getItem().getValue();
String prefix = CommonConstant.PREFIX_RULE;
String key = prefix + sourceType + "_" + sourceId + "_" + item;
Boolean hasKey = redisStreamUtil.hasKey(key);
if (!hasKey) return;
List<AlarmRule> alarmRules = (List<AlarmRule>) redisStreamUtil.get(key);
double value = warnDto.getValue();
for (AlarmRule alarmRule : alarmRules) {
String operator = alarmRule.getOperator();
if (StrUtil.isBlank(operator))continue;
ObjectMapper mapper = new ObjectMapper();
Rule rule = mapper.readValue(operator, Rule.class);
boolean needWarn = parse(rule, value);
// 报警信息发送方式
String notific = alarmRule.getNotification();
// 联系人组id
String contactId = alarmRule.getContactId();
// 发送预警信息
if (needWarn)sendMessage.send(contactId,notific,rule);
}
}
/**
* 规则解析
*
* @return 是否需要发送预警信息
*/
private boolean parse(Rule rule,double value){
String op = rule.getOperator();
double threshold = rule.getThreshold();
if (EQ.getOp().equals(op)){
return value == threshold;
} else if (GT.getOp().equals(op)) {
return value > threshold;
} else if (GE.getOp().equals(op)) {
return value >= threshold;
} else if (LT.getOp().equals(op)) {
return value < threshold;
} else if (LE.getOp().equals(op)) {
return value <= threshold;
}else {
return false;
}
}
private void init(){
sendMessage = SpringContextUtils.getBean(SendMessage.class);
redisStreamUtil = SpringContextUtils.getBean(RedisStreamUtil.class);
}
}

42
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/redisStream/ConsumeA2.java Normal file
View File

@ -0,0 +1,42 @@
package org.jeecg.modules.redisStream;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.util.RedisUtil;
import org.jeecg.modules.base.dto.WarnDto;
import org.jeecg.modules.base.enums.SourceType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
@Data
@Slf4j
@Component
@NoArgsConstructor
public class ConsumeA2 implements StreamListener<String, ObjectRecord<String, WarnDto>> {
private String consumerName;
@Autowired
private RedisUtil redisUtil;
@Override
public void onMessage(ObjectRecord<String,WarnDto> message) {
String streamKey = message.getStream();
RecordId recordId = message.getId();
WarnDto msg = message.getValue();
String prefix = CommonConstant.PREFIX_RULE;
SourceType sourceType = msg.getSourceType();
String sourceId = msg.getSourceId();
String key = prefix + sourceType + "_" + sourceId;
log.info("[自动ACK][name:"+consumerName+"][streamKey:{}][id:{}][message:{}]", streamKey, recordId, msg);
// redisStreamUtil.del(streamKey, id.getValue());
}
public ConsumeA2(String consumerName) {
this.consumerName = consumerName;
}
}

24
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/redisStream/ConsumeB1.java Normal file
View File

@ -0,0 +1,24 @@
package org.jeecg.modules.redisStream;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.common.util.RedisStreamUtil;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import java.util.Map;
@Slf4j
@Component
public class ConsumeB1 implements StreamListener<String, MapRecord<String, String, String>> {
private RedisStreamUtil redisStreamUtil;
@Override
public void onMessage(MapRecord<String, String, String> message) {
String streamKey = message.getStream();
RecordId id = message.getId();
Map<String, String> value = message.getValue();
log.info("[自动ACK][group:Group_Warn_B][streamKey:{}][id:{}][value:{}]", streamKey, id, value);
// redisStreamUtil.del(streamKey, id.getValue());
}
}

189
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/redisStream/RedisStreamConfig.java Normal file
View File

@ -0,0 +1,189 @@
package org.jeecg.modules.redisStream;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import org.jeecg.common.constant.CommonConstant;
import org.jeecg.common.exception.StreamErrorHandler;
import org.jeecg.common.util.RedisStreamUtil;
import org.jeecg.modules.base.dto.WarnDto;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.stream.StreamMessageListenerContainer.*;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
@Configuration
public class RedisStreamConfig {
// 空闲线程的存活时间(s)
private final Integer keepAliveTime = 5;
// 每次轮询取几条消息
private final Integer maxMsg = 10;
// Stream Key
private final String warnKey = CommonConstant.STREAM_KEY_WARN;
// 消费组名
private final String groupWarnA = CommonConstant.GROUP_WARN_A;
// 消费者名
private final String consumerWarnA1 = CommonConstant.CONSUMERWARNA1;
@Resource
private RedisConnectionFactory redisConnectionFactory;
@Resource
private RedisStreamUtil redisStreamUtil;
@Bean(initMethod = "start", destroyMethod = "stop")
public StreamMessageListenerContainer<String, ObjectRecord<String, WarnDto>> streamMessageListenerContainer() {
/* 创建Stream和消费组A */
creatGroup(warnKey, groupWarnA);
// 原子整数,多线程环境下对整数的原子性操作
AtomicInteger index = new AtomicInteger(1);
// 返回当前系统可用的处理器数量
Integer processors = Runtime.getRuntime().availableProcessors();
/*
corePoolSize：线程池中核心线程的数量,即线程池中保持的最小线程数
maximumPoolSize：线程池中允许的最大线程数
keepAliveTime：非核心线程的空闲时间超过该值,就会被回收
unit：keepAliveTime 参数的时间单位
workQueue：用于保存等待执行的任务的阻塞队列
threadFactory：用于创建新线程的工厂
**/
ThreadPoolExecutor executor = new ThreadPoolExecutor(processors,
processors,
keepAliveTime,
TimeUnit.SECONDS,
new LinkedBlockingDeque<>(),
r -> {
Thread thread = new Thread(r);
thread.setName("Thread-Stream-Consumer-" + index.getAndIncrement());
thread.setDaemon(true);
return thread;
});
/* 设置消息监听容器 */
StreamMessageListenerContainerOptions<String, ObjectRecord<String,WarnDto>> options =
StreamMessageListenerContainerOptions
.builder()
// 每次轮询取几条消息
.batchSize(maxMsg)
// Stream执行消息轮询的执行器
.executor(executor)
// Stream中没有消息时,阻塞多长时间(轮询等待时间)
// 设置为0表示消费者将一直等待新消息到达
// 不能大于spring.redis.timeout
.pollTimeout(Duration.ZERO)
//.serializer()
//.objectMapper(new ObjectHashMapper())
.targetType(WarnDto.class)
// 异常处理器
.errorHandler(new StreamErrorHandler())
.build();
/* 创建消息监听容器 */
StreamMessageListenerContainer<String, ObjectRecord<String,WarnDto>> streamMessageListenerContainer =
StreamMessageListenerContainer.create(redisConnectionFactory, options);
// 独立消费
/*streamMessageListenerContainer.receive(StreamOffset.fromStart(streamKey),
new ConsumeListener("独立消费", null, null));*/
/*
register:用于注册一个消息监听器,该监听器将在每次有新的消息到达Redis Stream时被调用
这种方式适用于长期运行的消息消费者，它会持续监听Redis Stream并处理新到达的消息
receive:用于主动从Redis Stream中获取消息并进行处理,你可以指定要获取的消息数量和超时时间
这种方式适用于需要主动控制消息获取的场景,例如批量处理消息或定时任务
**/
// 注册消费组A中的消费者A1,手动ACK
/*ConsumerStreamReadRequest<String> readA1 = StreamMessageListenerContainer
.StreamReadRequest
.builder(StreamOffset.create(warnKey, ReadOffset.lastConsumed()))
.consumer(Consumer.from(groupWarnA, consumerWarnA1))
.autoAcknowledge(false)
// 如果消费者发生了异常,是否禁止消费者消费
.cancelOnError(throwable -> false)
.build();
ConsumeA1 consumeA1 = new ConsumeA1(groupWarnA, consumerWarnA1);
streamMessageListenerContainer.register(readA1, consumeA1);*/
ConsumeA1 consumeA1 = new ConsumeA1(groupWarnA, consumerWarnA1);
streamMessageListenerContainer.receive(Consumer.from(groupWarnA, consumerWarnA1),
StreamOffset.create(warnKey, ReadOffset.lastConsumed()), consumeA1);
// 创建消费组A中的消费者A2,自动ACK
/* ConsumeA2 consumeA2 = new ConsumeA2(consumerWarnA2);
streamMessageListenerContainer.receiveAutoAck(Consumer.from(groupWarnA, consumerWarnA2),
StreamOffset.create(warnKey, ReadOffset.lastConsumed()), consumeA2);*/
// 注册消费组B中的消费者B1,自动ACK
/*streamMessageListenerContainer.receiveAutoAck(Consumer.from(groupWarnB, consumerWarnB1),
StreamOffset.create(warnKey, ReadOffset.lastConsumed()), new ConsumeListenerB());*/
return streamMessageListenerContainer;
}
/**
* 创建Stream、单个消费组
*/
private void creatGroup(String streamKey,String groupName){
if (StrUtil.isBlank(streamKey) || StrUtil.isBlank(groupName))return;
if (redisStreamUtil.hasKey(streamKey)) {
StreamInfo.XInfoGroups groups = redisStreamUtil.getGroups(streamKey);
if (groups.isEmpty()) {
redisStreamUtil.createGroup(streamKey, groupName);
}else {
// 判断该组是否已经创建
List<String> created = groups.stream()
.map(StreamInfo.XInfoGroup::groupName)
.collect(Collectors.toList());
if (!created.contains(groupName))redisStreamUtil.createGroup(streamKey, groupName);
}
} else {
redisStreamUtil.createGroup(streamKey, groupName);
}
}
/**
* 创建Stream、多个消费组
*/
private void creatGroup(String streamKey, List<String> groupNames){
if (StrUtil.isBlank(streamKey) || CollUtil.isEmpty(groupNames))return;
if (redisStreamUtil.hasKey(streamKey)) {
StreamInfo.XInfoGroups groups = redisStreamUtil.getGroups(streamKey);
if (groups.isEmpty()) {
groupNames.forEach(groupName -> {
redisStreamUtil.createGroup(streamKey, groupName);
});
}else {
// 如果组名已经存在,从待创建列表中移除
List<String> created = groups.stream()
.map(StreamInfo.XInfoGroup::groupName)
.collect(Collectors.toList());
Iterator<String> iterator = groupNames.iterator();
while (iterator.hasNext()){
String groupName = iterator.next();
if (created.contains(groupName))iterator.remove();
}
// 对不存在的组进行创建
groupNames.forEach(groupName -> {
redisStreamUtil.createGroup(streamKey, groupName);
});
}
} else {
groupNames.forEach(groupName -> {
redisStreamUtil.createGroup(streamKey, groupName);
});
}
}
}

1
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/system/controller/SysUserController.java
View File

@ -1765,5 +1765,4 @@ public class SysUserController {
public Map<String, SysUser> findUserMap(){
return sysUserService.findUserMap();
}
}

1
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/system/service/ISysUserService.java
View File

@ -404,4 +404,5 @@ public interface ISysUserService extends IService<SysUser> {
*/
Map<String, SysUser> findUserMap();
List<SysUser> listByIds(List<String> ids);
}

10
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/system/service/impl/SysBaseApiImpl.java
View File

@ -26,10 +26,7 @@ import org.jeecg.common.system.vo.*;
import org.jeecg.common.util.*;
import org.jeecg.common.util.dynamic.db.FreemarkerParseFactory;
import org.jeecg.modules.message.entity.SysMessageTemplate;
import org.jeecg.modules.message.handle.impl.DdSendMsgHandle;
import org.jeecg.modules.message.handle.impl.EmailSendMsgHandle;
import org.jeecg.modules.message.handle.impl.QywxSendMsgHandle;
import org.jeecg.modules.message.handle.impl.SystemSendMsgHandle;
import org.jeecg.modules.message.handle.impl.*;
import org.jeecg.modules.message.service.ISysMessageTemplateService;
import org.jeecg.modules.message.websocket.WebSocket;
import org.jeecg.modules.system.entity.*;
@ -1191,6 +1188,9 @@ public class SysBaseApiImpl implements ISysBaseAPI {
@Autowired
private DdSendMsgHandle ddSendMsgHandle;
@Autowired
private EmailPushMsgHandle emailPushMsgHandle;
@Override
public void sendTemplateMessage(MessageDTO message) {
String messageType = message.getType();
@ -1226,6 +1226,8 @@ public class SysBaseApiImpl implements ISysBaseAPI {
ddSendMsgHandle.sendMessage(message);
}else if(MessageTypeEnum.QYWX.getType().equals(messageType)){
qywxSendMsgHandle.sendMessage(message);
} else if (MessageTypeEnum.YX.getType().equals(messageType)) { // 新增邮件推送 2023-06-25
emailPushMsgHandle.sendMessage(message);
}
}

12
jeecg-module-system/jeecg-system-biz/src/main/java/org/jeecg/modules/system/service/impl/SysUserServiceImpl.java
View File

@ -1,5 +1,6 @@
package org.jeecg.modules.system.service.impl;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
@ -1329,6 +1330,17 @@ public class SysUserServiceImpl extends ServiceImpl<SysUserMapper, SysUser> impl
return map;
}
@Override
public List<SysUser> listByIds(List<String> ids) {
boolean isEmpty = CollUtil.isEmpty(ids);
if (isEmpty){
return new ArrayList<>();
}
LambdaQueryWrapper<SysUser> wrapper = new LambdaQueryWrapper<>();
wrapper.in(SysUser::getId,ids);
return list(wrapper);
}
/**
* 修改租户下的部门
* @param departs

12
jeecg-server-cloud/jeecg-system-cloud-start/pom.xml
View File

@ -12,18 +12,6 @@
<description>System项目微服务启动</description>
<dependencies>
<!-- 引入jeecg-boot-starter-cloud依赖 -->
<dependency>
<groupId>org.jeecgframework.boot</groupId>
<artifactId>jeecg-boot-starter-cloud</artifactId>
<!-- 3.2版本号后，可选择是否排除jeecg-system-cloud-api，不排除会优先通过fegin调用接口
<exclusions>
<exclusion>
<groupId>org.jeecgframework.boot</groupId>
<artifactId>jeecg-system-cloud-api</artifactId>
</exclusion>
</exclusions>-->
</dependency>
<!-- jeecg-system-biz依赖 -->
<dependency>
<groupId>org.jeecgframework.boot</groupId>