java多线程实现mq消息处理超时监控

 

项目运用rabbitMq 实现的分布式架构，每台机器既是生成者也是消费者，改项目依赖于外部调度服务，mq的consumer listener引用其他组的jar包（实现对外部资源的调用），消息消费是单线程的，在此调用jar包操作业务或者是个人代码写的有问题，可能导致线程死锁，或者其他代码问题（http请求未设置超时），导致消息消费被卡住，最终消息堆积。导致正常业务垮掉。既然consumer没配置消息的超时时间，java支持多线程，那么可以用来完美解决，worker线程执行任务，protect线程监听worker线程判断执行时间，（其实这两个线程都是在同时互相监听，各自执行完毕stop未执行完成的线程）不啰嗦直接上代码

 

 BaseThreadUtil类 worker 线程和protect线程集成该抽象类

/**
 * Created by liweigao on 2017/4/25.
 */
public abstract class BaseThreadUtil extends Thread {


  public abstract void execute();//抽象方法需要子类实现

  private String threadName = "";

  //在父类重写run方法，在子类只要重写execute方法就可以了
  @Override
  public void run() {
    super.run();
    execute();
  }

  //在需要回调数据的地方（两个子类需要），声明一个接口
  public static interface Callback {

    public void complete();
  }

  //2.创建接口对象
  public Callback callback;

  public String getThreadName() {
    return threadName;
  }

  public void setThreadName(String threadName) {
    this.threadName = threadName;
  }
}

 worker 线程类 执行主要业务线程。

/**
 * Created by liweigao on 2017/4/25.
 */
public class WorkerThread extends BaseThreadUtil {

  private Logger logger = LoggerFactory.getLogger(WorkerThread.class);

  private Runnable runnable;

  public WorkerThread(Runnable runnable, String threadName) {
    this.runnable = runnable;
    if (threadName != null && !"".equals(threadName)) {
      super.setThreadName(threadName);
    } else {
      super.setThreadName("worker thread");
    }
  }

  @Override
  public void execute() {
    StopWatch stopWatch=new StopWatch();
    stopWatch.start();
    if (runnable != null) {
      runnable.run();
    }
    stopWatch.stop();
//    System.out.println("线程：" + super.getThreadName() + "执行完毕，开始执行回调……");
    logger.debug("线程：" + super.getThreadName() + "执行完毕，开始执行回调……耗时："+stopWatch.getTotalTimeMillis() +"ms");
    //任务执行完毕  执行回调
    callback.complete();
  }
}

 protect 线程类。

 

/**
 * Created by liweigao on 2017/4/25.
 */
public class ProtectThread extends BaseThreadUtil {

  private Logger logger = LoggerFactory.getLogger(ProtectThread.class);

  private Integer timeout = 6000;

  public ProtectThread(Integer timeout, String threadName) {
    this.timeout = timeout;
    if (threadName != null && !"".equals(threadName)) {
      super.setThreadName(threadName);
    } else {
      super.setThreadName("protect thread");
    }
  }

  @Override
  public void execute() {
    StopWatch stopWatch = new StopWatch();
    stopWatch.start();
    try {
      Thread.sleep(timeout);
    } catch (InterruptedException e) {
      logger.error(e.getMessage(), e);

    }
    stopWatch.stop();
//    System.out.println("线程：" + super.getThreadName() + "执行完毕，开始执行回调……");
    logger.debug(
        "线程：" + super.getThreadName() + "执行完毕，开始执行回调……耗时：" + stopWatch.getTotalTimeMillis() + "ms");
    //线程任务执行完毕 执行回调
    callback.complete();
  }
}

 baseStopUtil 类 也可以叫回调类，处理未完成线程。

/**
 * Created by liweigao on 2017/4/25.
 */
public class BaseStopUtil implements Callback {

  private Logger logger = LoggerFactory.getLogger(BaseStopUtil.class);
  BaseThreadUtil baseCallBackUtil;

  // 获取对象
  public BaseStopUtil(BaseThreadUtil baseCallBackUtil) {
    this.baseCallBackUtil = baseCallBackUtil;
  }

  @Override
  public void complete() {
//    System.out.println("线程：" + baseCallBackUtil.getThreadName() + "被停掉……");
    logger.debug("线程：" + baseCallBackUtil.getThreadName() + "被停掉……");
    if (baseCallBackUtil.isAlive()) {
      baseCallBackUtil.stop();
    }
  }
}

 

ListenThreadConsumer 封装对外调用方法，启动两个线程，配置超时时间。

/**
 * Created by liweigao on 2017/4/25.
 */
public class ListenThreadConsumer {

  /**
   * 过期时间
   */
  private Integer timeout;

  /**
   * 任务
   */
  private Runnable runnable;

  /**
   * 检测间隔时间 默认1000
   */
//  private Integer spacetime;
  public ListenThreadConsumer(Integer timeout, Runnable runnable) {
    this.timeout = timeout;
    this.runnable = runnable;
  }

  public void execute() {

    ProtectThread protectThread = new ProtectThread(timeout, "");
    WorkerThread workerThread = new WorkerThread(runnable, "");
    protectThread.callback = new BaseStopUtil(workerThread);
    workerThread.callback = new BaseStopUtil(protectThread);
    protectThread.start();
    workerThread.start();
  }
}

 那么就可以很简单的运用到自己的程序中了。main方法测试。

  public static void main(String[] args) {
    ListenThreadConsumer listenThreadConsumer = new ListenThreadConsumer(100, new Runnable() {
      @Override
      public void run() {
        System.out.println("这是我的测试……………………");
      }
    });
    //执行任务以及监控
    listenThreadConsumer.execute();
  }

打印日志：
这是我的测试…………………… 
2017-04-28 18:56:50.916 [Thread-1] DEBUG c.w.s.c.util.thread.WorkerThread-[34] - 线程：worker thread执行完毕，开始执行回调……耗时：1ms 
2017-04-28 18:56:50.922 [Thread-1] DEBUG c.w.s.c.util.thread.BaseStopUtil-[23] - 线程：protect thread被停掉…… 

以上完毕，可直接copy复用。以上信息欢迎大神吐槽，欢迎提建议。互相监控，其中哪一个线程有问题，都会被停掉。