使用JSch远程部署flume采集点-阿里云开发者社区

公司有个需求，为了减少运维人员的工作量，需要开发一个远程部署flume的工具。我这里使用比较方便安全的SFTP协议，前提是需要客户端安装有SSH。Unix、linux、aix系统基本默认安装，windows的要自行安装。

其实在windows系统中还可以使用SMB文件共享协议来实现文件传输，还有比较通用的FTP协议、Telnet协议，但这里我们选用比较安全，操作方便的SFTP协议，其他的，我在下面也贴出来。

首先下载依赖jar包：jsch-0.1.52.jar。官网：http://www.jcraft.com/jsch/

然后编写工具类，具体看代码：

package com.mysite.stfp;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Map;
import java.util.Properties;
import java.util.Vector;
import org.apache.log4j.Logger;
import com.jcraft.jsch.Channel;
import com.jcraft.jsch.ChannelExec;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.ChannelSftp.LsEntry;
import com.jcraft.jsch.JSch;
import com.jcraft.jsch.JSchException;
import com.jcraft.jsch.Session;
import com.jcraft.jsch.SftpException;
public class SFTPChannel {
  private static final Logger LOG = Logger.getLogger(SFTPChannel.class.getName());
  private Session session = null;
  private ChannelSftp sftp = null;
  private ChannelExec exec = null;
  public SFTPChannel(String host, int port, String userName, String password, int timeout) {
    try {
      JSch jsch = new JSch(); // 创建JSch对象
      session = jsch.getSession(userName, host, port);// 根据用户名，主机ip，端口获取一个Session对象
      LOG.debug("Session created.");
      if (password != null) {
        session.setPassword(password); // 设置密码
      }
      Properties config = new Properties();
      config.put("StrictHostKeyChecking", "no");
      session.setConfig(config); // 为Session对象设置properties
      session.setTimeout(timeout); // 设置timeout时间
      session.connect(); // 通过Session建立链接
      LOG.debug("Session connected.");
    } catch (JSchException e) {
      e.printStackTrace();
    }
  }
  public SFTPChannel(Map<String, String> map, int timeout) {
    try {
      String host = map.get("host").toString();
      int port = Integer.parseInt(map.get("port") + "");
      String userName = map.get("userName").toString();
      String password = map.get("password").toString();
      JSch jsch = new JSch(); // 创建JSch对象
      session = jsch.getSession(userName, host, port);// 根据用户名，主机ip，端口获取一个Session对象
      LOG.debug("Session created.");
      if (password != null) {
        session.setPassword(password); // 设置密码
      }
      Properties config = new Properties();
      config.put("StrictHostKeyChecking", "no");
      session.setConfig(config); // 为Session对象设置properties
      session.setTimeout(timeout); // 设置timeout时间
      session.connect(); // 通过Session建立链接
      LOG.debug("Session connected.");
    } catch (JSchException e) {
      e.printStackTrace();
    }
  }
  /**
   * 打开SFTP通道
   */
  public ChannelSftp getSftp() {
    LOG.debug("Opening Channel.");
    try {
      Channel channel = session.openChannel("sftp");// 打开SFTP通道
      channel.connect(); // 建立SFTP通道的连接
      sftp = (ChannelSftp) channel;
    } catch (JSchException e) {
      e.printStackTrace();
    }
    return sftp;
  }
  /**
   * 打开exec通道
   */
  public ChannelExec getExec() {
    LOG.debug("Opening Channel.");
    try {
      Channel channel = session.openChannel("exec");// 打开SFTP通道
      channel.setInputStream(null);
      channel.connect(); // 建立SFTP通道的连接
      exec = (ChannelExec) channel;
    } catch (JSchException e) {
      e.printStackTrace();
    }
    return exec;
  }
  public void execCmd(String command) throws JSchException {
    BufferedReader reader = null;
    Channel channel = null;
    try {
      channel = session.openChannel("exec");
      ((ChannelExec) channel).setCommand(command);
      channel.setInputStream(null);
      ((ChannelExec) channel).setErrStream(System.err);
      channel.connect();
      reader = new BufferedReader(new InputStreamReader(channel.getInputStream()));
      String buf = null;
      while ((buf = reader.readLine()) != null) {
        System.out.println(buf);
      }
    } catch (IOException e) {
      e.printStackTrace();
    } catch (JSchException e) {
      e.printStackTrace();
    } finally {
      try {
        reader.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
      channel.disconnect();
    }
  }
  public void pwd() throws SftpException {
    LOG.info(sftp.pwd());
  }
  public void ls() throws SftpException {
    ls(".");
  }
  public void ls(String path) throws SftpException {
    Vector<?> vector = sftp.ls(path);
    for (Object object : vector) {
      if (object instanceof LsEntry) {
        LsEntry entry = LsEntry.class.cast(object);
        LOG.info(entry.getFilename());
      }
    }
  }
  public void closeChannel() {
    if (sftp != null){
      sftp.quit();
      sftp.disconnect();
    }
    if (exec != null)
      exec.disconnect();
    if (session != null) {
      session.disconnect();
    }
  }
  public void closeSftp() {
    if (sftp != null){
      sftp.quit();
      sftp.disconnect();
    }
  }
  public void closeExec() {
    if (exec != null)
      exec.disconnect();
  }
}

然后是调用方法：

@RequestMapping(value = "/sftp.do")
  public String uploadBySftp(HttpServletRequest request, HttpServletResponse response) {
    String result = "handle failed!";
    SFTPChannel channel = null;
    try {
      if (logger.isDebugEnabled()) {
        logger.debug("execute uploadBySftp method...");
      }
      Map<String, String> params = getParas(request);
      String path = params.get("path").toString();
      String os = params.get("os").toString();
      String userName = params.get("userName").toString();
      params.put("userName", decoder(userName));
      String password = params.get("password").toString();
      params.put("password", decoder(password));
      channel = new SFTPChannel(params, 60000);
      ChannelSftp chSftp = channel.getSftp();
      try {
        channel.ls(path);
      } catch (Exception e) {
        chSftp.mkdir(path);
      }
      File file = new File(src);
      long fileSize = file.length();
      chSftp.put(src, path, new FileProgressMonitor(fileSize), ChannelSftp.OVERWRITE);
      String fileName = file.getName();
      if (file.isFile() && (fileName.endsWith(".zip") || fileName.endsWith(".ZIP")))
        channel.execCmd("unzip -o " + path + fileName + " -d " + path);
      else if (file.isFile() && (fileName.endsWith(".tar") || fileName.endsWith(".tar.gz")))
        channel.execCmd("tar -zxvf " + path + fileName);
      chSftp.rm(path + file.getName());
      if (!"windows".equals(os)) {
        channel.execCmd(path + "flume/bin/flume.sh");
      } else {
        channel.execCmd(path + "flume\\bin\\start.bat");
      }
      result = "success";
      return result;
    } catch (SftpException e) {
      logger.error("has a error:{}", e.getMessage());
      e.printStackTrace();
      result = "file create failed";
      return result;
    } catch (JSchException e) {
      logger.error("has a error:{}", e.getMessage());
      e.printStackTrace();
      return result;
    } finally {
      if (channel != null)
        channel.closeChannel();
    }
  }

下面是重写的文件传输的百分比统计：

package com.mysite.stfp;
import java.text.DecimalFormat;
import java.util.Timer;
import java.util.TimerTask;
import com.jcraft.jsch.SftpProgressMonitor;
public class FileProgressMonitor extends TimerTask implements SftpProgressMonitor {
  private long progressInterval = 1000; // 默认间隔时间为1秒
  private boolean isEnd = false; // 记录传输是否结束
  private long transfered; // 记录已传输的数据总大小
  private long fileSize; // 记录文件总大小
  private Timer timer; // 定时器对象
  private boolean isScheduled = false; // 记录是否已启动timer记时器
  public FileProgressMonitor(long fileSize) {
    this.fileSize = fileSize;
  }
  @Override
  public void run() {
    if (!isEnd()) { // 判断传输是否已结束
      //System.out.println("Transfering is in progress.");
      long transfered = getTransfered();
      if (transfered != fileSize) { // 判断当前已传输数据大小是否等于文件总大小
        //System.out.println("Current transfered: " + transfered + " bytes");
        sendProgressMessage(transfered);
      } else {
        System.out.println("File transfering is done.");
        setEnd(true); // 如果当前已传输数据大小等于文件总大小，说明已完成，设置end
      }
    } else {
      //System.out.println("Transfering done. Cancel timer.");
      stop(); // 如果传输结束，停止timer记时器
      return;
    }
  }
  public void stop() {
    //System.out.println("Try to stop progress monitor.");
    if (timer != null) {
      timer.cancel();
      timer.purge();
      timer = null;
      isScheduled = false;
    }
    System.out.println("Progress monitor stoped.");
  }
  public void start() {
    //System.out.println("Try to start progress monitor.");
    if (timer == null) {
      timer = new Timer();
    }
    timer.schedule(this, 1000, progressInterval);
    isScheduled = true;
    System.out.println("Progress monitor started.");
  }
  /**
   * 打印progress信息
   * 
   * @param transfered
   */
  private void sendProgressMessage(long transfered) {
    if (fileSize != 0) {
      double d = ((double) transfered * 100) / (double) fileSize;
      DecimalFormat df = new DecimalFormat("#.##");
      System.out.println("Sending progress message: " + df.format(d) + "%");
    } else {
      System.out.println("Sending progress message: " + transfered);
    }
  }
  /**
   * 实现了SftpProgressMonitor接口的count方法
   */
  public boolean count(long count) {
    if (isEnd())
      return false;
    if (!isScheduled) {
      start();
    }
    add(count);
    return true;
  }
  /**
   * 实现了SftpProgressMonitor接口的end方法
   */
  public void end() {
    setEnd(true);
    System.out.println("transfering end.");
  }
  private synchronized void add(long count) {
    transfered = transfered + count;
  }
  private synchronized long getTransfered() {
    return transfered;
  }
  public synchronized void setTransfered(long transfered) {
    this.transfered = transfered;
  }
  private synchronized void setEnd(boolean isEnd) {
    this.isEnd = isEnd;
  }
  private synchronized boolean isEnd() {
    return isEnd;
  }
  public void init(int op, String src, String dest, long max) {
    // Not used for putting InputStream
  }
}

------------------------------------------------------------------------------------------------

以上是SFTP协议的一些方法，下面介绍smb协议操作，依赖的包是jcifs-1.3.18.jar：

package com.mysite.smb;
import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import jcifs.smb.SmbFile;
import jcifs.smb.SmbFileInputStream;
import jcifs.smb.SmbFileOutputStream;
/**
 * 
 * @author T430
 * 
 */
public class SmbUtil {
  /**
   * 方法一：
   * 
   * @param remoteUrl
   *            远程路径 smb://192.168.75.204/test/新建 文本文档.txt
   * @throws IOException
   */
  public static void smbGet(String remoteUrl) throws IOException {
    SmbFileInputStream in = null;
    try {
      SmbFile smbFile = new SmbFile(remoteUrl);
      int length = smbFile.getContentLength();// 得到文件的大小
      byte buffer[] = new byte[length];
      in = new SmbFileInputStream(smbFile);
      // 建立smb文件输入流
      while ((in.read(buffer)) != -1) {
        System.out.write(buffer);
        System.out.println(buffer.length);
      }
    } catch (IOException e) {
      throw new IOException(e);
    } finally {
      try {
        if (in != null)
          in.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }
  // 从共享目录下载文件
  /**
   * 方法二： 路径格式：smb://192.168.75.204/test/新建 文本文档.txt
   * smb://username:password@192.168.0.77/test
   * 
   * @param remoteUrl
   *            远程路径
   * @param localDir
   *            要写入的本地路径
   * @throws Exception
   */
  public static void smbGet(String remoteUrl, String localDir) throws IOException {
    InputStream in = null;
    OutputStream out = null;
    try {
      SmbFile remoteFile = new SmbFile(remoteUrl);
      String fileName = remoteFile.getName();
      File localFile = new File(localDir + File.separator + fileName);
      in = new BufferedInputStream(new SmbFileInputStream(remoteFile));
      out = new BufferedOutputStream(new FileOutputStream(localFile));
      byte[] buffer = new byte[1024];
      while (in.read(buffer) != -1) {
        out.write(buffer);
        buffer = new byte[1024];
      }
    } catch (IOException e) {
      throw new IOException(e);
    } finally {
      try {
        if (out != null)
          out.close();
        if (in != null)
          in.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }
  /**
   * 向共享目录上传文件 远程url smb://192.168.1.77/test 如果需要用户名密码就这样：
   * smb://username:password@192.168.1.77/test
   * 
   * @param remoteUrl
   * @param localFilePath
   * @throws Exception
   */
  public static void smbPut(String remoteUrl, String localFilePath) throws IOException {
    InputStream in = null;
    OutputStream out = null;
    try {
      File localFile = new File(localFilePath);
      String fileName = localFile.getName();
      SmbFile remoteFile = new SmbFile(remoteUrl + "/" + fileName);
      in = new BufferedInputStream(new FileInputStream(localFile));
      out = new BufferedOutputStream(new SmbFileOutputStream(remoteFile));
      byte[] buffer = new byte[1024];
      while (in.read(buffer) != -1) {
        out.write(buffer);
        buffer = new byte[1024];
      }
    } catch (IOException e) {
      throw new IOException(e);
    } finally {
      try {
        if (out != null)
          out.close();
        if (in != null)
          in.close();
      } catch (IOException e) {
        e.printStackTrace();
      }
    }
  }
}

FTP的操作，依赖包commons-net-3.3.jar：

package com.mysite.ftp;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Map;
import org.apache.commons.io.IOUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
public class FtpUtil {
  private static FTPClient ftp;
  /**
   * FTP上传
   * 
   * @param map
   * @param src
   */
  public static void upload(Map<String, Object> map, String src) {
    String host = map.get("host").toString();
    String userName = map.get("userName").toString();
    String password = map.get("password").toString();
    String path = map.get("path") == null ? "" : map.get("path").toString();
    String port = map.get("port") == null ? "21" : map.get("port").toString();
    upload(host, Integer.parseInt(port), userName, password, path, src);
  }
  /**
   * FTP上传
   * 
   * @param host
   * @param userName
   * @param password
   * @param path
   * @param src
   */
  public static void upload(String host, int port, String userName, String password, String path, String src) {
    try {
      boolean isConnect = connect(path, host, port, userName, password);
      if(isConnect){
        File file = new File(src);
        upload(file);
      }
    } catch (IOException e) {
      e.printStackTrace();
      throw new RuntimeException("FTP客户端出错！", e);
    } finally {
      try {
        ftp.disconnect();
      } catch (IOException e) {
        e.printStackTrace();
        throw new RuntimeException("关闭FTP连接发生异常！", e);
      }
    }
  }
  /**
   * 
   * @param path
   *            上传到ftp服务器哪个路径下
   * @param addr
   *            地址
   * @param port
   *            端口号
   * @param username
   *            用户名
   * @param password
   *            密码
   * @return
   * @throws Exception
   */
  private static boolean connect(String path, String addr, int port, String username, String password) throws IOException {
    boolean result = false;
    ftp = new FTPClient();
    int reply;
    ftp.connect(addr, port);
    ftp.login(username, password);
    ftp.setFileType(FTPClient.BINARY_FILE_TYPE);
    reply = ftp.getReplyCode();
    if (!FTPReply.isPositiveCompletion(reply)) {
      ftp.disconnect();
      return result;
    }
    ftp.changeWorkingDirectory(path);
    ftp.setBufferSize(1024);
    ftp.setControlEncoding("GBK");
    result = true;
    return result;
  }
  /**
   * 
   * @param file
   *            上传的文件或文件夹
   * @throws Exception
   */
  private static void upload(File file) throws IOException {
    if (file.isDirectory()) {
      ftp.makeDirectory(file.getName());
      ftp.changeWorkingDirectory(file.getName());
      String[] files = file.list();
      for (int i = 0; i < files.length; i++) {
        File file1 = new File(file.getPath() + "\\" + files[i]);
        if (file1.isDirectory()) {
          upload(file1);
          ftp.changeToParentDirectory();
        } else {
          File file2 = new File(file.getPath() + "\\" + files[i]);
          FileInputStream input = new FileInputStream(file2);
          ftp.storeFile(file2.getName(), input);
          input.close();
        }
      }
    } else {
      File file2 = new File(file.getPath());
      FileInputStream input = new FileInputStream(file2);
      ftp.storeFile(file2.getName(), input);
      input.close();
    }
  }
  /**
   * FTP下载
   * 
   * @param host
   * @param userName
   * @param password
   * @param path
   * @param src
   */
  public static void download(String host, String userName, String password, String path, String src) {
    FTPClient ftpClient = new FTPClient();
    FileOutputStream fos = null;
    try {
      ftpClient.connect(host);
      ftpClient.login(userName, password);
      String remoteFileName = path;
      fos = new FileOutputStream(src);
      ftpClient.setBufferSize(1024);
      // 设置文件类型（二进制）
      ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE);
      ftpClient.retrieveFile(remoteFileName, fos);
    } catch (IOException e) {
      e.printStackTrace();
      throw new RuntimeException("FTP客户端出错！", e);
    } finally {
      IOUtils.closeQuietly(fos);
      try {
        ftpClient.disconnect();
      } catch (IOException e) {
        e.printStackTrace();
        throw new RuntimeException("关闭FTP连接发生异常！", e);
      }
    }
  }
}

Telnet协议，依赖包commons-net-3.3.jar：

package com.mysite.telnet;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.net.SocketException;
import org.apache.commons.net.telnet.TelnetClient;
public class TelnetUtil {
  private static InputStream in;
  private static PrintStream out;
  private static String prompt = "#";
  private static String os = "windows";
  public static void handle(String host, int port, String os, String user, String pwd) {
    TelnetClient telnet = null;
    try {
      if ("windows".equals(os)) {
        telnet = new TelnetClient("VT220");
        prompt = "";
      } else {
        TelnetUtil.os = os;
        telnet = new TelnetClient();
        prompt = user.equals("root") ? "#" : "$";
      }
      telnet.connect(host, port);
      telnet.setSoTimeout(60000);
      in = telnet.getInputStream();
      out = new PrintStream(telnet.getOutputStream());
      login(user, pwd);
      //sendCommand("cd /d \"D:/ftp/flume/bin\"");
      //sendCommand("start \"\" \"D:/ftp/flume/bin/startCWAgen.bat\"");
      sendCommand("java -jar \"D:/ftp/flume/lib/controller.jar\" \"D:/ftp/flume/lib/config.properties\"");
      readUntil();
    } catch (SocketException e) {
      e.printStackTrace();
    } catch (IOException e) {
      e.printStackTrace();
    } finally {
      try {
        if (telnet != null)
          telnet.disconnect();
      } catch (IOException e) {
        e.printStackTrace();
        try {
          throw new Exception("telnet 关闭失败");
        } catch (Exception e1) {
          e1.printStackTrace();
        }
      }
    }
  }
  public static void handle(String host, String os, String user, String pwd) {
    int port = 23;
    handle(host, port, os, user, pwd);
  }
  private static void login(String user, String password) {
    readUntil("login:");
    write(user);
    if ("windows".equals(os)) {
      readUntil("password:");
    } else {
      readUntil("Password:");
    }
    write(password);
    readUntil(prompt + " ");
  }
  private static void su(String password) {
    try {
      write("su");
      readUntil("Password: ");
      write(password);
      prompt = "#";
      readUntil(prompt + " ");
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  private static void write(String value) {
    try {
      out.println(value);
      out.flush();
      System.out.println(value);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }
  public static String sendCommand(String command) {
    try {
      write(command);
      return readUntil(prompt + " ");
    } catch (Exception e) {
      e.printStackTrace();
    }
    return null;
  }
  private static String readUntil() {
    InputStreamReader isr = null;
    BufferedReader br = null;
    try {
      //StringBuffer sb = new StringBuffer();
      isr = new InputStreamReader(in, "GBK");
      br = new BufferedReader(isr);
      int str = 0;
      while ((str = br.read()) != -1) {
        //sb.append((char) str);
        System.out.print((char) str);
      }
    } catch (Exception e) {
      e.printStackTrace();
    } finally {
      try {
        if (isr != null)
          isr.close();
        if (br != null)
          br.close();
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
    return null;
  }
  private static String readUntil(String pattern) {
    StringBuffer sb = new StringBuffer();
    try {
      char lastChar = pattern.charAt(pattern.length() - 1);
      char ch = (char) in.read();
      while (true) {
        sb.append(ch);
        if (ch == lastChar) {
          if (sb.toString().endsWith(pattern)) {
            byte[] temp = sb.toString().getBytes("iso8859-1");// 处理编码，界面显示乱码问题
            return new String(temp, "GBK");
          }
        }
        ch = (char) in.read();
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
    return sb.toString();
  }
  public static void main(String[] args) {
    TelnetUtil.handle("127.0.0.1", "windows", "bob", "123");
  }
}

使用JSch远程部署flume采集点

热门文章

最新文章

相关课程

相关电子书