Java多线程编程模式实战指南之Promise模式

public class DataSyncTask implements Runnable {
    private final Map<String, String> taskParameters;
    public DataSyncTask(Map<String, String> taskParameters) {
        this.taskParameters = taskParameters;
    }
    @Override
    public void run() {
        String ftpServer = taskParameters.get("server");
        String ftpUserName = taskParameters.get("userName");
        String password = taskParameters.get("password");
        
        //先开始初始化FTP客户端实例
        Future<FTPClientUtil> ftpClientUtilPromise = FTPClientUtil.newInstance(
            ftpServer, ftpUserName, password);
        //查询数据库生成本地文件
        generateFilesFromDB();
        FTPClientUtil ftpClientUtil = null;
        try {
            // 获取初始化完毕的FTP客户端实例
            ftpClientUtil = ftpClientUtilPromise.get();
        } catch (InterruptedException e) {
            ;
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
        // 上传文件
        uploadFiles(ftpClientUtil);
        //省略其他代码
    }
    private void generateFilesFromDB() {
        // 省略其他代码
    }
    private void uploadFiles(FTPClientUtil ftpClientUtil) {
        Set<File> files = retrieveGeneratedFiles();
        for (File file : files) {
            try {
                ftpClientUtil.upload(file);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    private Set<File> retrieveGeneratedFiles() {
        Set<File> files = new HashSet<File>();
        // 省略其他代码
        return files;
    }
}

//模式角色：Promise.Promisor、Promise.Result
public class FTPClientUtil {
    private final FTPClient ftp = new FTPClient();

    private final Map<String, Boolean> dirCreateMap = new HashMap<String, Boolean>();

    private FTPClientUtil() {

    }

    //模式角色：Promise.Promisor.compute
    public static Future<FTPClientUtil> newInstance(final String ftpServer,
        final String userName, final String password) {

        Callable<FTPClientUtil> callable = new Callable<FTPClientUtil>() {

            @Override
            public FTPClientUtil call() throws Exception {
                FTPClientUtil self = new FTPClientUtil();
                self.init(ftpServer, userName, password);
                return self;
            }

        };

        //task相当于模式角色：Promise.Promise
        final FutureTask<FTPClientUtil> task = new FutureTask<FTPClientUtil>(
            callable);

        /*
        下面这行代码与本案例的实际代码并不一致，这是为了讨论方便。
        下面新建的线程相当于模式角色：Promise.TaskExecutor
        */
        new Thread(task).start();
        return task;
    }

    private void init(String ftpServer, String userName, String password)
        throws Exception {

        FTPClientConfig config = new FTPClientConfig();
        ftp.configure(config);

        int reply;
        ftp.connect(ftpServer);

        System.out.print(ftp.getReplyString());

        reply = ftp.getReplyCode();

        if (!FTPReply.isPositiveCompletion(reply)) {
            ftp.disconnect();
            throw new RuntimeException("FTP server refused connection.");
        }
        boolean isOK = ftp.login(userName, password);
        if (isOK) {
            System.out.println(ftp.getReplyString());

        } else {
            throw new RuntimeException("Failed to login." + ftp.getReplyString());
        }

        reply = ftp.cwd("~/subspsync");
        if (!FTPReply.isPositiveCompletion(reply)) {
            ftp.disconnect();
            throw new RuntimeException("Failed to change working directory.reply:"
                + reply);
        } else {

            System.out.println(ftp.getReplyString());
        }

        ftp.setFileType(FTP.ASCII_FILE_TYPE);

    }

    public void upload(File file) throws Exception {
        InputStream dataIn = new BufferedInputStream(new FileInputStream(file),
            1024 * 8);
        boolean isOK;
        String dirName = file.getParentFile().getName();
        String fileName = dirName + ‘/‘ + file.getName();
        ByteArrayInputStream checkFileInputStream = new ByteArrayInputStream(
            "".getBytes());

        try {
            if (!dirCreateMap.containsKey(dirName)) {
                ftp.makeDirectory(dirName);
                dirCreateMap.put(dirName, null);
            }

            try {
                isOK = ftp.storeFile(fileName, dataIn);
            } catch (IOException e) {
                throw new RuntimeException("Failed to upload " + file, e);
            }
            if (isOK) {
                ftp.storeFile(fileName + ".c", checkFileInputStream);

            } else {

                throw new RuntimeException("Failed to upload " + file + ",reply:" + 
","+ ftp.getReplyString());
            }
        } finally {
            dataIn.close();
        }

    }

    public void disconnect() {
        if (ftp.isConnected()) {
            try {
                ftp.disconnect();
            } catch (IOException ioe) {
                // 什么也不做
            }
        }
    }
}

public class FTPClientUtil {
    private volatilestatic ThreadPoolExecutor threadPoolExecutor;

    static {
        threadPoolExecutor = new ThreadPoolExecutor(1,Runtime.getRuntime()
            .availableProcessors() * 2, 
            60, 
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<Runnable>(10), new ThreadFactory() {
                public Thread newThread(Runnable r) {
                    Thread t = new Thread(r);
                    t.setDaemon(true);
                    return t;
                }
            }, new ThreadPoolExecutor.CallerRunsPolicy());
    }
    
    private final FTPClient ftp = new FTPClient();
    private final Map<String, Boolean> dirCreateMap = new HashMap<String, Boolean>();

    //私有构造器
    private FTPClientUtil() {

    }

    public static Future<FTPClientUtil> newInstance(final String ftpServer,
        final String userName, final String password) {

        Callable<FTPClientUtil> callable = new Callable<FTPClientUtil>() {

            @Override
            public FTPClientUtil call() throws Exception {
                FTPClientUtil self = new FTPClientUtil();
                self.init(ftpServer, userName, password);
                return self;
            }

        };
        final FutureTask<FTPClientUtil> task = new FutureTask<FTPClientUtil>(
            callable);

        threadPoolExecutor.execute(task);
        return task;
    }

    private void init(String ftpServer, String userName, String password)
        throws Exception {
        //省略与清单6-2中相同的代码
    }

    public void upload(File file) throws Exception {
        //省略与清单6-2中相同的代码
    }

    public void disconnect() {
        //省略与清单6-2中相同的代码
    }

}