使用JSch远程部署flume采集点

2653885 · 发表于 2015-11-27 20:03:35

　　公司有个需求，为了减少运维人员的工作量，需要开发一个远程部署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.coreware.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-daemon.sh");
} else {
channel.execCmd(path + "flume\\bin\\startCWAgen.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.coreware.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.coreware.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.0.77/test 如果需要用户名密码就这样：
* smb://username:password@192.168.0.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.coreware.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);
if (file1.isDirectory()) {
upload(file1);
ftp.changeToParentDirectory();
} else {
File file2 = new File(file.getPath() + "\\" + files);
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.coreware.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");
}
}

　　

账号		自动登录	找回密码
密码			立即注册

VMware vcenter+vSphere 6.5 U2共享

【跟谁学】韩宇极简英语课-技术人员不得不

用Zabbix通过JMX方式监控weblogic

winhex数据恢复教程（非常巨大，内容丰富）

Symantec Backup Exec 2015 2016/2012 BE20

NetScaler VPX部署之：NetScaler Gateway调

zabbix3.4.1安装部署+微信推送信息+大屏显

[经验分享] 使用JSch远程部署flume采集点

扫码加入运维网微信交流群