远程执行shell命令

 

 

 

import com.jcraft.jsch.*;
import com.xx.dc.beans.Progress;
import com.xx.dc.beans.ProgressMessage;
import com.xx.dc.service.SessionManager;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.*;
import java.util.Arrays;
import java.util.Date;
import java.util.List;


public class ShellUtils {
    private static Log log = LogFactory.getLog(ShellUtils.class);


    public static Session getSession(String user, String passwd, String host) throws JSchException {

        JSch jsch = new JSch();
        Session session = jsch.getSession(user, host, 22);
        session.setPassword(passwd);

        java.util.Properties config = new java.util.Properties();
        config.put("StrictHostKeyChecking", "no");
        session.setConfig(config);
        session.connect();
        return session;
    }

    /**
     * 执行相关的命令
     * @throws JSchException
     */
    public static void execCmd(String command, String user, String passwd, String host, final String processID) throws JSchException {
        execCmd(command, user, passwd, host, processID, null);
    }

    public static void execCmd(Session session, String command, final String processID, PrintStream ps) throws JSchException {
        StringBuilder ret = new StringBuilder();

        BufferedReader reader = null;
        Channel channel = null;

        try {
            channel = session.openChannel("exec");
            ((ChannelExec) channel).setCommand(command);

            channel.setInputStream(null);
            ((ChannelExec) channel).setErrStream(new ErrorOutputStream(new IWriter() {
                @Override
                public void write(String content) {
                        trace(processID, content);
                }

                @Override
                public void write(String content, Exception e) {
                        trace(processID, content, e);
                }
            }));

            channel.connect();
            InputStream in = channel.getInputStream();
            reader = new BufferedReader(new InputStreamReader(in));
            String buf = null;
            while ((buf = reader.readLine()) != null) {
                if (ps != null) {
                    ps.println(buf);
                }
                    trace(processID, buf);
            }
        } catch (IOException e) {
            trace(processID, e.getMessage(), e);
        } catch (JSchException e) {
            trace(processID, e.getMessage(), e);
        } finally {
            try {
                channel.disconnect();
                reader.close();
            } catch (IOException e) {
                trace(processID, e.getMessage(), e);
            }
        }
    }

    /**
     * 执行相关的命令
     * @throws JSchException
     */
    public static void execCmd(String command, String user, String passwd, String host, final String processID, PrintStream ps) throws JSchException {

        Session session = getSession(user, passwd, host);
        try {
            execCmd(session, command, processID, ps);
        } finally {
            session.disconnect();
        }

    }

    private static void trace(String processID, String message) {
        trace(processID, message, null);
    }

    private static void trace(String processID, String message, Throwable e) {
        MemCachedManager memCachedManager = MemCachedManager.getInstance();
        Object progress = memCachedManager.get(processID);
        if (progress == null) {
            progress = new Progress(processID);
        }
        List<ProgressMessage> messages = ((Progress) progress).getMessages();

        if (e == null) {
            log.info("[Shell Processing:" + processID + "]" + message);
            messages.add(new ProgressMessage(message));
        } else {
            messages.add(new ProgressMessage(message, true));
            if (!StringUtils.isEmpty(e.getMessage())) {
                messages.add(new ProgressMessage("Error details:" + e.getMessage(), true));
                log.error("[Shell Processing:" + processID + "]" + message, e);

            } else {
                log.error("[Shell Processing:" + processID + "]" + message);
            }
            ((Progress) progress).setHasError(true);
        }


        Date expiry = new Date(System.currentTimeMillis() + (1000 * 60 * 60));// 有效期一个小时
        memCachedManager.set(processID, progress, expiry);
    }

    public static void main(String[] args) {
        try {

            execCmd("hadoop distcp -overwrite hdfs://172.18.149.42:50070/tmp/dmp/ode/all_click_stream_data hdfs://yh/dmp/external/ode/all_click_stream_data", "dmp", "1234qwer", "172.18.149.134", "test");
//System.out.println(System.getProperty("line.separator"));

        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}