Home  >  Article  >  Java  >  Java multi-threading implements Callable interface

Java multi-threading implements Callable interface

高洛峰
高洛峰Original
2017-01-05 15:04:271393browse

Calling method:

/**
 * 点击量/月(年)Callable
 */
 public void yearlyClickCallable() {
 // 获取参数
 String year = getPara("year");
 // 统计数据集X
 List<String> xList = new ArrayList<String>();
 xList.add("January");
 xList.add("February");
 xList.add("March");
 xList.add("April");
 xList.add("May");
 xList.add("June");
 xList.add("July");
 xList.add("August");
 xList.add("September");
 xList.add("October");
 xList.add("November");
 xList.add("December");
 // 统计数据集Y
 List<Integer> yList = new ArrayList<Integer>();
 // 接收线程值
 List<Future<List<Map<String, Object>>>> futureList = new ArrayList<Future<List<Map<String, Object>>>>();
 // 计数器
 int count = 0;
 // 创建一个线程池(决定开启几个线程)
 ExecutorService pool = Executors.newCachedThreadPool();
 // 每月的日志分析
 for (int m = 1; m <= 12; m++) {
  // 收集日期参数
  List<String> dateList = new ArrayList<String>();
  //
  String date = "";
  // 判断有多少天
  int days = CalendarUtil.weekForMonth(Integer.valueOf(year), m);
  // 组合日期
  for (int i = 1; i <= days; i++) {
 
  if (i <= 9) {
 
   if (m <= 9) {
   date = year + "-0" + m + "-0" + i;
   } else {
   date = year + "-" + m + "-0" + i;
   }
  } else {
   if (m <= 9) {
   date = year + "-0" + m + "-" + i;
   } else {
   date = year + "-" + m + "-" + i;
   }
  }
  dateList.add(date);
  }
  // 启动
  Future<List<Map<String, Object>>> future = pool.submit(new ReadLogFileCallableByYear(dateList));
 
  futureList.add(future);
 }
 // 关闭线程池
 pool.shutdown();
 // 接收结果集
 for (Future<List<Map<String, Object>>> future : futureList) {
  try {
  // 接收参数
  List<Map<String, Object>> list = future.get(1, TimeUnit.SECONDS);
 
  // 设置参数
  for (int p = 0; p < list.size(); p++) {
 
   count += (int) list.get(p).get("clickCount");
 
   if (list.get(p).get("month").equals("01")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("02")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("03")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("04")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("05")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("06")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("07")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("08")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("09")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("10")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("11")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   } else if (list.get(p).get("month").equals("12")) {
   yList.add((Integer) list.get(p).get("clickCount"));
   }
 
  }
  } catch (Exception e) {
  e.printStackTrace();
  }
 }
 
 setAttr("totalCount", count);
 setAttr("x", xList);
 setAttr("y", yList);
 renderJson();
 }

Multi-threading method:

package com.ninemax.util.loganalysis;
 
import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
 
import com.ninemax.util.loganalysis.tool.ConstantUtil;
 
/**
 * 多线程有返回值
 * 
 * @author Darker
 * 
 */
public class ReadLogFileCallableByYear implements Callable<List<Map<String, Object>>> {
 // 日期数组
 private List<String> clickDate;
 // 返回结果集
 public List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
 
 public ReadLogFileCallableByYear(List<String> clickDate) {
 this.clickDate = clickDate;
 }
 
 @Override
 public List<Map<String, Object>> call() throws Exception {
 // 接收参数
 Map<String, Object> map = new HashMap<String, Object>();
 // 利用FileInputStream读取文件信息
 FileInputStream fis = null;
 // 利用InputStreamReader进行转码
 InputStreamReader reader = null;
 // 利用BufferedReader进行缓冲
 BufferedReader bufReader = null;
 // 利用StringBuffer接收文件内容容器
 StringBuffer buf = new StringBuffer();
 // 点击量/月
 int monthClick = 0;
 
 for (int i = 0; i < clickDate.size(); i++) {
  // 获取文件
  File clickLogFile = new File(ConstantUtil.LOGLOCATION, "article.click."+ clickDate.get(i) + ".txt");
  // 判断文件是否存在
  if (!clickLogFile.exists() || clickLogFile.isDirectory()) {
 
  System.err.println(clickDate.get(i) + "的文件不存在...");
   
  map.put("month", clickDate.get(i).substring(5, 7));
  map.put("clickCount", 0);
  list.add(map);
   
  return list;
  } else {
  try {
   // 节点流
   fis = new FileInputStream(clickLogFile);
   // 转换流
   reader = new InputStreamReader(fis, "utf-8");
   // 处理流
   bufReader = new BufferedReader(reader);
   // 计数器
   int count = 0;
   // 按行读取
   String line = "";
   // 读取文件
   while ((line = bufReader.readLine()) != null) {
   // 计数
   count++;
   // 接收数据
   if (!line.equals(null) && !line.equals("")) {
 
    buf.append(line + "\n");
   }
   }
   if (count == 0) {
   count = 0;
   } else {
   count = count - 1;
   }
   monthClick += count;
  } catch (Exception e) {
   e.printStackTrace();
  } finally {
   // 关闭流
   try {
   bufReader.close();
   reader.close();
   fis.close();
   } catch (IOException e) {
   e.printStackTrace();
   }
  }
  }
 }
 // 结果集
 map.put("month", clickDate.get(0).substring(5, 7));
  
 if (monthClick == 0) {
  map.put("clickCount", 0);
 } else {
  map.put("clickCount", monthClick);
 }
 
 list.add(map);
 
 return list;
 }
 
}

Let me share with you an example from a netizen, which is also very good

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
 
/**
 * Callable 和 Future接口
 * Callable是类似于Runnable的接口,实现Callable接口的类和实现Runnable的类都是可被其它线程执行的任务。
 * Callable和Runnable有几点不同: 
 * (1)Callable规定的方法是call(),而Runnable规定的方法是run().
 * (2)Callable的任务执行后可返回值,而Runnable的任务是不能返回值的。 
 * (3)call()方法可抛出异常,而run()方法是不能抛出异常的。
 * (4)运行Callable任务可拿到一个Future对象, Future表示异步计算的结果。
 * 它提供了检查计算是否完成的方法,以等待计算的完成,并检索计算的结果。
 * 通过Future对象可了解任务执行情况,可取消任务的执行,还可获取任务执行的结果。
 */
public class CallableAndFuture {
 
    /**
     * 自定义一个任务类,实现Callable接口
     */
    public static class MyCallableClass implements Callable {
        // 标志位
        private int flag = 0;
 
        public MyCallableClass(int flag) {
            this.flag = flag;
        }
 
        public String call() throws Exception {
            if (this.flag == 0) {
                // 如果flag的值为0,则立即返回
                return "flag = 0";
            }
            if (this.flag == 1) {
                // 如果flag的值为1,做一个无限循环
                try {
                    while (true) {
                        System.out.println("looping......");
                        Thread.sleep(2000);
                    }
                } catch (InterruptedException e) {
                    System.out.println("Interrupted");
                }
                return "false";
            } else {
                // falg不为0或者1,则抛出异常
                throw new Exception("Bad flag value!");
            }
        }
    }
 
    public static void main(String[] args) {
        // 定义3个Callable类型的任务
        MyCallableClass task1 = new MyCallableClass(0);
        MyCallableClass task2 = new MyCallableClass(1);
        MyCallableClass task3 = new MyCallableClass(2);
 
        // 创建一个执行任务的服务
        ExecutorService es = Executors.newFixedThreadPool(3);
        try {
            // 提交并执行任务,任务启动时返回了一个Future对象,
            // 如果想得到任务执行的结果或者是异常可对这个Future对象进行操作
            Future future1 = es.submit(task1);
            // 获得第一个任务的结果,如果调用get方法,当前线程会等待任务执行完毕后才往下执行
            System.out.println("task1: " + future1.get());
 
            Future future2 = es.submit(task2);
            // 等待5秒后,再停止第二个任务。因为第二个任务进行的是无限循环
            Thread.sleep(5000);
            System.out.println("task2 cancel: " + future2.cancel(true));
 
            // 获取第三个任务的输出,因为执行第三个任务会引起异常
            // 所以下面的语句将引起异常的抛出
            Future future3 = es.submit(task3);
            System.out.println("task3: " + future3.get());
        } catch (Exception e) {
            System.out.println(e.toString());
        }
        // 停止任务执行服务
        es.shutdownNow();
    }
}

More Java multi-threading implementations For articles related to Callable interface, please pay attention to the PHP Chinese website!

Statement:
The content of this article is voluntarily contributed by netizens, and the copyright belongs to the original author. This site does not assume corresponding legal responsibility. If you find any content suspected of plagiarism or infringement, please contact admin@php.cn