线程并发编程Callable与Future_JAVA_编程开发_程序员俱乐部

//1-使用5个线程并发-Runnable
?private void send(List<Integer> list){
??List<DataPO> files = dataService.findByIds(list);
??files = Collections.synchronizedList(files);
??for(int i=0;i<5;i++){
???new Thread(new SendDataThread(files)).start();
??}
?}
?
?//2-使用2个线程并发-Callable+FutureTask: 使用Callable+FutureTask获取执行结果
?private void send2(List<Integer> list){
??List<DataPO> files = dataService.findByIds(list);
??//files = Collections.synchronizedList(files);
??//分隔数据
??int len = files.size();
??List<DataPO> files1 = files.subList(0, len/2);
??List<DataPO> files2 = files.subList(len/2, len);
??Map<String, List<DataPO>> map = new HashMap<String, List<DataPO>>();
??map.put("file0", files1);
??map.put("file1", files2);
??Callable[] calls = new Callable[2];
??FutureTask[] tasks = new FutureTask[2];
??try {
???for(int i=0;i<2;i++){
????calls[i] = new SendDataThread2(map.get("file" + i));
????tasks[i] = new FutureTask(calls[i]);
???}
???
???for(int i=0;i<2;i++){
????new Thread(tasks[i]).start();
???}
???System.out.println("**");
???
???List<Integer> result = null;
???for(int i=0;i<2;i++){
????result = (List<Integer>)tasks[i].get();
????System.out.println("*** result " + result.size());
???}
???System.out.println("*****");
??} catch (Exception e) {
???e.printStackTrace();
??}
?}
?
?//3-使用2个线程并发-Callable+Future: Callable+Future获取执行结果
?private void send3(List<Integer> list){
??List<DataPO> files = dataService.findByIds(list);
??int len = files.size();
??List<DataPO> files1 = files.subList(0, len/2);
??List<DataPO> files2 = files.subList(len/2, len);
??Map<String, List<DataPO>> map = new HashMap<String, List<DataPO>>();
??map.put("file0", files1);
??map.put("file1", files2);
//??files = Collections.synchronizedList(files);
??Callable[] calls = new Callable[2];
??for(int i=0;i<2;i++){
???calls[i] = new SendDataThread3(map.get("file" + i));
??}
??// 创建一个执行任务的服务
??ExecutorService es = Executors.newFixedThreadPool(2);
??try { // 提交并执行任务，任务启动时返回了一个Future对象，
???// 如果想得到任务执行的结果或者是异常可对这个Future对象进行操作
???Future future1 = es.submit(calls[0]);
???System.out.println("*");
???// 获得第一个任务的结果，如果调用get方法，当前线程会等待任务执行完毕后才往下执行
???//System.out.println("task1: " + future1.get());
???Future future2 = es.submit(calls[1]);
???System.out.println("**");
???// 等待5秒后，再停止第二个任务。因为第二个任务进行的是无限循环
???//Thread.sleep(5000);
???System.out.println("task1: " + future1.get());
???System.out.println("***");
???System.out.println("task2: " + future2.get());
???//System.out.println("task2 cancel: " + future2.cancel(true));
??} catch (Exception e) { System.out.println(e.toString()); }
??// 停止任务执行服务
??System.out.println("*****");
??files.clear();
??es.shutdownNow();
?}

?

public class SendDataThread3 implements Callable<List<Integer>> {
?private List<DataPO> files;
?
?public SendDataThread3(List<DataPO> files){
??this.files = files;
?}

?public List<Integer> call() throws Exception {
??List<Integer> result = new ArrayList<Integer>();
??DataPO file = null;
??int len = files.size();
??int no = 0;
??while(no<len){
???file = files.get(no++);
//???files.remove(0);
???result.add(file.getId());
???Thread.currentThread().sleep(1000);
???System.out.println(Thread.currentThread().getName() + "??? " + file.getSyscod() + ", " + file.getFilename());
??}
??return result;
?}

}

?