package lwc;

import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;

import java.io.IOException;
import java.io.InputStream;

/**
 * Created by lwc on 16-2-19.
 */
public class TailThread implements Runnable {
    /**
     * 遠端 執行命令並返回結果呼叫過程 是同步的（執行完才會返回）
     *
     * @param host    主機名
     * @param user    使用者名稱
     * @param psw     密碼
     * @param port    埠
     * @param command 命令
     * @return
     */
    String host;
    String user;
    String psw;
    int port;
    String command;
    String interestStr;
    TailThread(String host, String user, String psw, int port, String command, String interestStr) {
        this.host = host;
        this.user = user;
        this.psw =  psw;
        this.port = port;
        this.command = command;
        this.interestStr = interestStr;
    }
    public void run() {
        this.exec(host, user, psw, port, command, interestStr);
    }

    /**
     * 遠端 執行命令並返回結果呼叫過程 是同步的（執行完才會返回）
     *
     * @param host    主機名
     * @param user    使用者名稱
     * @param psw     密碼
     * @param port    埠
     * @param command 命令
     * @return
     */
    public String exec(String host, String user, String psw, int port, String command, String interestingStr) {
        byte[] tmp = new byte[1024]; //讀資料快取
        StringBuffer strBuffer = new StringBuffer();  //執行SSH返回的結果
        Session session = null;
        ChannelExec openChannel = null;
        try {
            JSch jsch = new JSch();
            session = jsch.getSession(user, host, port);
            java.util.Properties config = new java.util.Properties();
            config.put("StrictHostKeyChecking", "no");
            session.setConfig(config);
            session.setPassword(psw);
            session.connect();
            openChannel = (ChannelExec) session.openChannel("exec");

            ChannelExec ssh = openChannel;

            //返回的結果可能是標準資訊,也可能是錯誤資訊,所以兩種輸出都要獲取
            //一般情況下只會有一種輸出.
            //但並不是說錯誤資訊就是執行命令出錯的資訊,如獲得遠端java JDK版本就以
            //ErrStream來獲得.
            InputStream InputStream = ssh.getInputStream();
            InputStream ErrStream = ssh.getErrStream();

            ssh.setCommand(command);
            ssh.connect();

            //開始獲得SSH命令的結果
            while (true) {
                //獲得錯誤輸出
                while (ErrStream.available() > 0) {
                    int i = ErrStream.read(tmp, 0, 1024);
                    if (i < 0) {
                        break;
                    }

                    strBuffer.append(new String(tmp, 0, i));
//                    System.out.println("Error" + strBuffer.toString());
                    if(strBuffer.toString().contains(interestingStr)) {
                        LogListener.flag = true;
                        System.out.println(this.toString() + "&&&" + strBuffer.toString());
                        Thread.interrupted();
                    }
                    strBuffer = new StringBuffer();
                }

                //獲得標準輸出
                while (InputStream.available() > 0) {
                    int i = InputStream.read(tmp, 0, 1024);

                    if (i < 0) {
                        break;
                    }
                    strBuffer.append(new String(tmp, 0, i));
//                    System.out.println("Correct;" + strBuffer.toString());
                    if(strBuffer.toString().contains(interestingStr)) {
                        LogListener.flag = true;
                        System.out.println(this.toString() + "&&&" + strBuffer.toString());
                        Thread.interrupted();
                    }
                    strBuffer = new StringBuffer();
                }

                if (ssh.isClosed()) {
//                    System.out.println("exit-status: " + ssh.getExitStatus());
                    break;
                }
                try {
                    Thread.sleep(100);
                } catch (Exception ee) {
                }
            }
            return strBuffer.toString();

        } catch (JSchException | IOException e) {
            System.out.println(e.getMessage());
        } finally {
            if (openChannel != null && !openChannel.isClosed()) {
                openChannel.disconnect();
            }
            if (session != null && session.isConnected()) {
                session.disconnect();
            }
        }
        return strBuffer.toString();
    }

    @Override
    public String toString() {
        return "TailThread{" +
            "host='" + host + '\'' +
            ", interestStr='" + interestStr + '\'' +
            '}';
    }

}
 

<pre name="code" class="java">package lwc;

/**
 * Created by lwc on 16-2-19.
 */
public class LogListener {
  public static boolean flag = false;
  public int hostNum = 4; //default
  public String[] hosts = new String[hostNum];
  public int[] port = new int[hostNum];
  public String path[] = new String[hostNum];
  public String[] usr = new String[hostNum];
  public String[] pwd = new String[hostNum];
  public String interestingString = "Exception";

  public LogListener() {
    defaultInit();
  }

  public LogListener(int hostNum, String[] hosts, int[] port, String[] path, String[] usr, String[] pwd, String interestingString) {
    this.hostNum = hostNum;
    this.hosts = hosts;
    this.port = port;
    this.path = path;
    this.usr = usr;
    this.pwd = pwd;
    this.interestingString = interestingString;
  }

  public int getHostNum() {
    return hostNum;
  }

  public String[] getHosts() {
    return hosts;
  }

  public int[] getPort() {
    return port;
  }

  public String[] getPath() {
    return path;
  }

  public String[] getUsr() {
    return usr;
  }

  public String[] getPwd() {
    return pwd;
  }

  public String getInterestingString() {
    return interestingString;
  }

  public void setHostNum(int hostNum) {
    this.hostNum = hostNum;
  }

  public void setHosts(String[] hosts) {
    this.hosts = hosts;
  }

  public void setPort(int[] port) {
    this.port = port;
  }

  public void setPath(String[] path) {
    this.path = path;
  }

  public void setUsr(String[] usr) {
    this.usr = usr;
  }

  public void setPwd(String[] pwd) {
    this.pwd = pwd;
  }

  public void setInterestingString(String interestingString) {
    this.interestingString = interestingString;
  }

  public void defaultInit() {
    hostNum = 4;
    for (int i = 0; i < hostNum; i++) {
      hosts[i] = "172.168.256." +  i);
      port[i] = 22;
      path[i] = "/var/log/hadoop/namenode-server-yarn.log";
      usr[i] = "root";
      pwd[i] = "password";
    }
    interestingString = "Exception";
  }

  Thread[] threads = new Thread[hostNum];

  public void start() {
    for (int i = 0; i < hostNum; i++) {
      threads[i] = new Thread(new TailThread(hosts[i], usr[i], pwd[i], port[i], "tail -f " + path[i], interestingString));
      threads[i].start();
    }
//    while (true) {
////      System.out.println(flag);
//      if (flag) {
//        for(int i = 0; i < hostNum; i++) {
//          if(threads[i].isAlive()) {
//            threads[i].interrupt();
//            threads[i].stop();
//          }
//        }
//        break;
//      } else {
//        try {
//          Thread.sleep(2000);
//        } catch (InterruptedException e) {
//          e.printStackTrace();
//        }
//      }
//    }
  }

  public boolean pause() {
    for (int i = 0; i < hostNum; i++) {
      if (threads[i].isAlive()) {
        threads[i].interrupt();
        threads[i].stop();
      }
    }
    return flag;
  }
}
 

package lwc;


/**
 * Created by lwc on 16-2-18.
 */


public class SSHHelper {

    public static void main(String args[]) {
        LogListener logListener = new LogListener();
        logListener.setInterestingString("OK3r5454tt");
        logListener.start();
        System.out.println("123456");
        //run sql
        try {
            Thread.sleep(5000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println(logListener.pause());
    }
}