MysqlKiller 一个没有感情的mysql 进程杀手

MysqlKiller.java

package org.zeusro;

import com.google.gson.Gson;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeusro.util.SQLHelper;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.concurrent.*;

/**
 * 不停执行 kill 语句
 * <p/>
 * Created in 2018.08.18
 * <p/>
 *
 * @author <a href="https://zeusro.github.io/" style="background: #55a7e3;">Zeusro</a>
 */
public class MysqlKiller {

    static final Logger logger = LoggerFactory.getLogger(MysqlKiller.class);
    ExecutorService executor;
    private String dbConnectionString;
    private Date missionStartDate;
    private Date missionExpiredDate;

    /**
     * 相对过期时间
     *
     * @param dbConnectionString the db connection string
     */
    public MysqlKiller(String dbConnectionString) {
        this.dbConnectionString = dbConnectionString;
        init();
        //给它30分钟的时间执行
        setMissionMinute(30);
    }

    /**
     * 设置绝对过期,到点了自动掐断
     *
     * @param dbConnectionString the db connection string
     * @param missionExpiredDate the mission expired date
     */
    public MysqlKiller(String dbConnectionString, Date missionExpiredDate) {
        this.dbConnectionString = dbConnectionString;
        init();
        setMissionExpiredDate(missionExpiredDate);
    }

    private void executeKillOrder(List<String> orders) {
        if (orders == null || orders.size() == 0) {
            return;
        }
        orders.forEach((kill) -> {
            logger.info("即将执行SQL:" + kill);
        });
        SQLHelper.executeSQLs(dbConnectionString, orders);
    }

    /**
     * 获取需要执行的语句
     * </br>
     *
     * @return list
     */
    List<String> getKillOrder() {
        List<String> killOrders = new ArrayList<>();
        String sql = "SELECT \n" +
                "    CONCAT('kill ', thread_id, ';') as killtext\n" +
                "FROM\n" +
                "    (SELECT DISTINCT\n" +
                "        (i.trx_mysql_thread_id) thread_id\n" +
                "    FROM\n" +
                "        information_schema.innodb_trx i, (SELECT \n" +
                "        id, time\n" +
                "    FROM\n" +
                "        information_schema.processlist\n" +
                "    WHERE\n" +
                "        time = (SELECT \n" +
                "                MAX(time)\n" +
                "            FROM\n" +
                "                information_schema.processlist\n" +
                "            WHERE\n" +
                "                state = 'Waiting for table metadata lock'\n" +
                "                    AND SUBSTRING(info, 1, 5) IN ('alter' , 'optim', 'repai', 'lock ', 'drop ', 'creat'))) p\n" +
                "    WHERE\n" +
                "        TIMESTAMPDIFF(SECOND, i.trx_started, NOW()) > p.time\n" +
                "            AND i.trx_mysql_thread_id NOT IN (CONNECTION_ID() , p.id)) t;";
        try (Connection connection = DriverManager.getConnection(dbConnectionString);
             Statement statement = connection.createStatement()) {
            Class.forName("com.mysql.cj.jdbc.Driver");
            ResultSet rs = statement.executeQuery(sql);
            while (rs.next()) {
                String killOrder = rs.getString("killtext");
                if (null != killOrder && !killOrder.trim().isEmpty()) {
                    killOrders.add(killOrder);
                }
            }
        } catch (
                Exception e) {
            logger.warn(LoggerConfig.LOG_TEMPLATE, e.getClass().getName(), e.getMessage(), null, new Gson().toJson(e), "%n");
        }
        return killOrders;
    }

    private void init() {
        Date now = new Date();
        missionStartDate = now;
        logger.info("开始执行时间:" + missionStartDate);
    }

    /**
     * 设置强制过期时间
     *
     * @param missionStartDate the mission start date
     */
    public void setMissionExpiredDate(Date missionStartDate) {
        this.missionExpiredDate = missionStartDate;
        logger.info("预期终止时间:" + missionStartDate);
    }

    /**
     * 根据missionStartDate偏移
     *
     * @param minutes the minutes
     */
    public void setMissionMinute(int minutes) {
        if (minutes <= 0) {
            throw new IllegalArgumentException("seconds有误");
        }
        Calendar c = Calendar.getInstance();
        c.setTime(missionStartDate);
        //给它30分钟的时间执行
        c.add(Calendar.MINUTE, minutes);
        missionExpiredDate = c.getTime();
        logger.info("预期终止时间:" + c.getTime());
    }

    /**
     * 执行 kill 操作
     */
    public void startKilling() {
        logger.info("开始执行 kill 操作");
        executor = new ThreadPoolExecutor(
                1,
                2,
                0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(1024),
                new ThreadPoolExecutor.AbortPolicy());
        ExecutorCompletionService completionService = new ExecutorCompletionService(executor);
        completionService.submit(() -> {
            while (new Date().before(missionExpiredDate)) {
                //每五秒运行一次,防止读取数据库过频
                Thread.sleep(5000);
                List<String> orders = getKillOrder();
                logger.info("每五秒运行一次,防止读取数据库过频");
                executeKillOrder(orders);
            }
            logger.info("执行完毕");
            return true;
        });
    }

    public void stopKilling() {
        if (executor == null) {
            return;
        }
        executor.shutdownNow();

    }
}