【集成自动发送邮件】自动扫描已经完成的日志，发送邮件

byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobScheduler.java

 package com.byit.conf;
 
+import com.byit.thread.EmailScanHelper;
 import com.byit.thread.JobScheduleHelper;
 import com.byit.thread.LogScanHelper;
 import com.byit.thread.RunRecordingScanHelper;
@@ -21,12 +22,14 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
     private final JobScheduleHelper jobScheduleHelper;
     private final LogScanHelper logScanHelper;
     private final RunRecordingScanHelper runRecordingScanHelper;
+    private final EmailScanHelper emailScanHelper;
 
     @Autowired
-    public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper, RunRecordingScanHelper runRecordingScanHelper) {
+    public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper, RunRecordingScanHelper runRecordingScanHelper, EmailScanHelper emailScanHelper) {
         this.jobScheduleHelper = jobScheduleHelper;
         this.logScanHelper = logScanHelper;
         this.runRecordingScanHelper = runRecordingScanHelper;
+        this.emailScanHelper = emailScanHelper;
     }
 
     /**
@@ -38,6 +41,7 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
         this.jobScheduleHelper.doStop();
         this.logScanHelper.doStop();
         this.runRecordingScanHelper.doStop();
+        this.emailScanHelper.doStop();
     }
 
     /**
@@ -50,6 +54,6 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
         this.jobScheduleHelper.start();
         this.logScanHelper.start();
         this.runRecordingScanHelper.start();
-
+        this.emailScanHelper.start();
     }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/mapper/EmailAlarmMapper.java

@@ -13,6 +13,11 @@ import java.util.List;
 @Repository
 public interface EmailAlarmMapper {
     /**
+     * 查询没有告警的邮箱
+     * @return
+     */
+    List<EmailAlarm> findEmailAlarmByAlarmResult();
+    /**
      * 根据id查询邮件
      * @param id
      * @return

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/EmailAlarm.java

@@ -56,12 +56,6 @@ public class EmailAlarm implements Serializable {
     private String alarmEmail;
 
     /**
-     * 当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)
-     */
-    @ApiModelProperty("当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)")
-    private String alarmAction;
-
-    /**
      * 告警结果(0, 未告警  1,告警成功  2,告警失败)
      */
     @ApiModelProperty("告警结果(0, 未告警  1,告警成功  2,告警失败)")

byit-myth-core/myth-admin-core/src/main/java/com/byit/model/vo/EmailAlarmVo.java

@@ -71,12 +71,6 @@ public class EmailAlarmVo implements Serializable {
     private String alarmEmail;
 
     /**
-     * 当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)
-     */
-    @ApiModelProperty("当前工作流版本的告警的时机(0 不告警， 1 完成时告警， 2 失败时告警， 3 成功时告警)")
-    private String alarmAction;
-
-    /**
      * 告警结果(0, 未告警  1,告警成功  2,告警失败)
      */
     @ApiModelProperty("告警结果(0, 未告警  1,告警成功  2,告警失败)")

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/EmailAlarmService.java

@@ -14,6 +14,12 @@ import java.util.List;
  **/
 public interface EmailAlarmService {
     /**
+     * 查询没有告警的邮箱
+     * @return
+     */
+    List<EmailAlarm> findEmailAlarmByAlarmResult();
+
+    /**
      * 根据id查询邮件
      * @param id
      * @return

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/EmailAlarmServiceImpl.java

@@ -39,6 +39,11 @@ public class EmailAlarmServiceImpl implements EmailAlarmService {
     }
 
     @Override
+    public List<EmailAlarm> findEmailAlarmByAlarmResult() {
+        return emailAlarmMapper.findEmailAlarmByAlarmResult();
+    }
+
+    @Override
     public EmailAlarm findEmailAlarmById(Integer id) {
         return emailAlarmMapper.findEmailAlarmById(id);
     }
@@ -64,9 +69,6 @@ public class EmailAlarmServiceImpl implements EmailAlarmService {
     @Override
     @DataValidation
     public void sendEmail(@ParamValidation EmailAlarmVo record) {
-        record.setSendTime(new Date());
-        record.setAlarmTitle("测试邮件");
-        record.setAlarmContent("<div style='color:#F00;font-size:1000px'>测试邮件</div>");
         try{
             //发送一个支持复杂类型的邮件
             MimeMessageHelper messageHelper = new MimeMessageHelper(javaMailSender.createMimeMessage(), true);
@@ -97,9 +99,11 @@ public class EmailAlarmServiceImpl implements EmailAlarmService {
             throw new BusinessException(AdminEnums.SEND_EMAIL_FAILURE);
         }
         EmailAlarm emailAlarm = new EmailAlarm( );
+
         BeanUtils.copyProperties(record,emailAlarm);
+        emailAlarm.setSendTime(new Date());
         //保存邮件
-        saveEmailAlarm(emailAlarm);
+        updateEmailAlarm(emailAlarm);
     }
 
     @Override

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/EmailScanHelper.java

+package com.byit.thread;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.model.EmailAlarm;
+import com.byit.model.vo.EmailAlarmVo;
+import com.byit.service.EmailAlarmService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @program: com.byit.thread.EmailScanHelper
+ * @description: 对于邮箱的扫描
+ * @author: huangfu
+ * @date: 2020年1月7日10:44:01
+ **/
+@Component
+@Slf4j
+public class EmailScanHelper {
+
+    private final DataSource dataSource;
+    private final EmailAlarmService emailAlarmService;
+
+    private volatile boolean emailThreadIsStop = false;
+    private Thread emailThread;
+
+    public EmailScanHelper(DataSource dataSource, EmailAlarmService emailAlarmService) {
+        this.dataSource = dataSource;
+        this.emailAlarmService = emailAlarmService;
+    }
+
+
+    public void start(){
+        startScanNotSentEmailFlow();
+    }
+
+    public void startScanNotSentEmailFlow(){
+        emailThread = new Thread(() ->{
+            dateAligned(5000);
+            log.info("--------------【com.byit.thread.EmailScanHelper.startScanNotSentEmailFlow】init success---------------");
+            while (!emailThreadIsStop){
+                Connection conn = null;
+                Boolean connAutoCommit = null;
+                PreparedStatement preparedStatement = null;
+                try {
+                    /**
+                     * 添加行锁
+                     */
+                    conn = dataSource.getConnection();
+                    connAutoCommit = conn.getAutoCommit();
+                    conn.setAutoCommit(false);
+                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'emali_alarm_lock' FOR UPDATE ");
+                    preparedStatement.execute();
+                    //查询未告警的邮箱
+                    List<EmailAlarm> emailAlarmByAlarmResults = emailAlarmService.findEmailAlarmByAlarmResult();
+                    if(CollectionUtil.isNotEmpty(emailAlarmByAlarmResults)){
+                        emailAlarmByAlarmResults.forEach(emailAlarm ->{
+                            EmailAlarmVo emailAlarmVo = new EmailAlarmVo();
+                            BeanUtils.copyProperties(emailAlarm,emailAlarmVo);
+                            emailAlarmService.sendEmail(emailAlarmVo);
+                        });
+                    }else{
+                        dateAligned(20000);
+                    }
+                }catch (Exception e){
+                    e.printStackTrace();
+                }finally {
+                    //释放资源
+                    if(conn != null){
+                        try {
+                            conn.commit();
+                        } catch (SQLException e) {
+                            if(!emailThreadIsStop){
+                                log.error("--------------------【提交行锁出错】---------------------");
+                            }
+                        }
+                    }
+
+                    try {
+                        if(conn != null){
+                            conn.setAutoCommit(connAutoCommit);
+                        }
+                    } catch (SQLException e) {
+                        if(!emailThreadIsStop){
+                            log.error("--------------------【恢复自动提交出错】---------------------");
+                        }
+                    }
+
+                    if(preparedStatement != null){
+                        try {
+                            preparedStatement.close();
+                        } catch (SQLException e) {
+                            if(!emailThreadIsStop){
+                                log.error("--------------------【关闭执行器出错】---------------------");
+                            }
+                        }
+                    }
+
+                    try {
+                        conn.close();
+                    } catch (SQLException e) {
+                        if(!emailThreadIsStop){
+                            log.error("--------------------【关闭链接出错】---------------------");
+                        }
+                    }
+                }
+            }
+        });
+        emailThread.setDaemon(true);
+        emailThread.setName("myth-job#【EmailScanHelper】#startScanNotSentEmailFlow");
+        emailThread.start();
+    }
+
+    public void doStop(){
+        this.emailThreadIsStop = true;
+        try {
+            TimeUnit.SECONDS.sleep(1);
+        } catch (InterruptedException e) {
+            e.printStackTrace( );
+        }
+        if (emailThread.getState() != Thread.State.TERMINATED) {
+            emailThread.interrupt();
+
+            try {
+                emailThread.join();
+            } catch (InterruptedException e) {
+                e.printStackTrace( );
+            }
+        }
+        log.warn("---------------【日志扫描线程被注销】-----------------------");
+    }
+
+    /**
+     * 对齐时钟。整秒运行
+     */
+    private void dateAligned(long waitTime){
+        try {
+            TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
+        } catch (InterruptedException e) {
+            log.warn("----------------【线程被中断】-----------------------");
+        }
+    }
+
+}

byit-myth-core/myth-admin-core/src/main/resources/mapper/EmailAlarmMapper.xml

@@ -8,7 +8,6 @@
     <result column="run_id" jdbcType="VARCHAR" property="runId" />
     <result column="version_name" jdbcType="VARCHAR" property="versionName" />
     <result column="alarm_email" jdbcType="VARCHAR" property="alarmEmail" />
-    <result column="alarm_action" jdbcType="CHAR" property="alarmAction" />
     <result column="alarm_result" jdbcType="CHAR" property="alarmResult" />
     <result column="send_time" jdbcType="TIMESTAMP" property="sendTime" />
     <result column="alarm_title" jdbcType="VARCHAR" property="alarmTitle" />
@@ -18,12 +17,23 @@
     <result column="alarm_content" jdbcType="LONGVARCHAR" property="alarmContent" />
   </resultMap>
   <sql id="Base_Column_List">
-    id, flow_id, flow_name, run_id, version_name, alarm_email, alarm_action, alarm_result,
+    id, flow_id, flow_name, run_id, version_name, alarm_email, alarm_result,
     send_time,alarm_title,flow_res
   </sql>
   <sql id="Blob_Column_List">
     alarm_content
   </sql>
+
+  <select id="findEmailAlarmByAlarmResult" resultMap="ResultMapWithBLOBs">
+    select
+      <include refid="Base_Column_List" />
+      ,
+      <include refid="Blob_Column_List" />
+    from email_alarm
+    where alarm_result = '0'
+
+  </select>
+
   <select id="findEmailAlarmById" parameterType="java.lang.Integer" resultMap="ResultMapWithBLOBs">
     select
     <include refid="Base_Column_List" />
@@ -57,9 +67,6 @@
       <if test="alarmEmail != null">
         alarm_email,
       </if>
-      <if test="alarmAction != null">
-        alarm_action,
-      </if>
       <if test="alarmResult != null">
         alarm_result,
       </if>
@@ -95,9 +102,7 @@
       <if test="alarmEmail != null">
         #{alarmEmail,jdbcType=VARCHAR},
       </if>
-      <if test="alarmAction != null">
-        #{alarmAction,jdbcType=CHAR},
-      </if>
+
       <if test="alarmResult != null">
         #{alarmResult,jdbcType=CHAR},
       </if>
@@ -118,7 +123,7 @@
 
   <insert id="saveEmailAlarms" parameterType="com.byit.model.EmailAlarm">
         insert into email_alarm (
-            id, flow_id, flow_name, run_id, version_name, alarm_email, alarm_action, alarm_result,
+            id, flow_id, flow_name, run_id, version_name, alarm_email, alarm_result,
             send_time,alarm_content,alarm_title,flow_res
         ) values
     <foreach collection="emailAlarms" item="emailAlarm" separator =",">
@@ -126,9 +131,9 @@
         #{emailAlarm.id,jdbcType=INTEGER},#{emailAlarm.flowId,jdbcType=INTEGER},
         #{emailAlarm.flowName,jdbcType=VARCHAR},#{emailAlarm.runId,jdbcType=VARCHAR},
         #{emailAlarm.versionName,jdbcType=VARCHAR},#{emailAlarm.alarmEmail,jdbcType=VARCHAR},
-        #{emailAlarm.alarmAction,jdbcType=CHAR},#{emailAlarm.alarmResult,jdbcType=CHAR},
-        #{emailAlarm.sendTime,jdbcType=TIMESTAMP},#{emailAlarm.alarmContent,jdbcType=LONGVARCHAR},
-        #{emailAlarm.alarmTitle,jdbcType=VARCHAR},#{emailAlarm.flowRes,jdbcType=CHAR}
+        #{emailAlarm.alarmResult,jdbcType=CHAR},#{emailAlarm.sendTime,jdbcType=TIMESTAMP},
+        #{emailAlarm.alarmContent,jdbcType=LONGVARCHAR},#{emailAlarm.alarmTitle,jdbcType=VARCHAR},
+        #{emailAlarm.flowRes,jdbcType=CHAR}
       )
     </foreach>
   </insert>
@@ -151,9 +156,7 @@
       <if test="alarmEmail != null">
         alarm_email = #{alarmEmail,jdbcType=VARCHAR},
       </if>
-      <if test="alarmAction != null">
-        alarm_action = #{alarmAction,jdbcType=CHAR},
-      </if>
+
       <if test="alarmResult != null">
         alarm_result = #{alarmResult,jdbcType=CHAR},
       </if>

byit-myth-executor/myth-exector-plugin/src/main/java/com/byit/utils/JobUtils.java

@@ -25,7 +25,9 @@ import java.util.concurrent.ConcurrentHashMap;
 public class JobUtils {
     private static final String REQUEST_PREFIX = "http://";
     private static final String REQUEST_ADD_JOB_RESOURCES_SUFFIX = "/job/addJob";
-    //发布工作流
+    /**
+     * 发布工作流
+     */
     private static final String REQUEST_FLOW_PUBLISH = "/api/flow/publish";
     //创建工作空间
     private static final String REQUEST_WORKSPACE_ADD = "/api/workspace/add";

demo-client/byit-demo-client/src/main/java/com/byit/test/ExploringSynchronized.java

-package com.byit.test;
-
-/**
- * @author huangfu
- */
-public class ExploringSynchronized implements Runnable {
-    private static final String LOCK_MARK = "LOCK_MARK";
-    /**
-     * 共享资源(临界资源)
-     */
-    static int i=0;
-    public void add(){
-        synchronized (LOCK_MARK){
-            i++;
-        }
-    }
-
-    @Override
-    public void run() {
-        for (int j = 0; j < 100000; j++) {
-            add();
-        }
-    }
-
-    public static void main(String[] args) throws InterruptedException {
-        Thread t1 = new Thread(new ExploringSynchronized());
-        Thread t2 = new Thread(new ExploringSynchronized());
-        t1.start();
-        t2.start();
-        //join 主线程需要等待子线程完成后在结束
-        t1.join();
-        t2.join();
-        System.out.println(i);
-
-    }
-}