| ページ一覧 | ブログ | twitter |　 書式 | 書式(表) |

MyMemoWiki

Java 並行処理

提供: MyMemoWiki
2020年2月16日 (日) 04:27時点におけるPiroto (トーク | 投稿記録)による版
(差分) ← 古い版 | 最新版 (差分) | 新しい版 → (差分)
ナビゲーションに移動 検索に移動

Java 並行処理

Java |

非同期に grep を実施する例1

内容

  • Thread ではなく、Executor を使って、Runnable を非同期に実行する
  • 共有データにスレッドセーフな単一の変数(java.util.concurrent.atomic)を使用する

java.util.concurrent.atomic

  • 単一の変数に対するロックフリーでスレッドセーフなプログラミングをサポートするクラスの小規模なツールキット

Executor

  • java.util.concurrent
  • 送信された Runnable タスクを実行するオブジェクト
  • 通常、executor は、明示的にスレッドを作成する代わりに使用

ソース

  1. import java.io.BufferedReader;
  2. import java.io.BufferedWriter;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.FileOutputStream;
  6. import java.io.FilenameFilter;
  7. import java.io.InputStreamReader;
  8. import java.io.OutputStreamWriter;
  9. import java.util.ArrayList;
  10. import java.util.List;
  11. import java.util.concurrent.Executor;
  12. import java.util.concurrent.atomic.AtomicInteger;
  13.  
  14. /**
  15.  * 
  16.  */
  17. public class JGrep {
  18.  
  19.     // 同期された変数
  20.     //volatile int workingThreadCount;
  21.     static AtomicInteger workingThreadCount = null;
  22.     
  23.     private static final int THREAD_SIZE = 10;
  24.     public static void main(String[] args) throws Exception {
  25.     
  26.         if (args.length <=2) {
  27.             System.out.println("Usage: java JGrep 対象ディレクトリ  正規表現  [ファイルフィルタ]");
  28.             System.exit(-1);
  29.         }
  30.  
  31.         String fileFilter = "";
  32.         if (args.length > 2) {
  33.             fileFilter = args[2];
  34.         }
  35.         (new JGrep()).grep(args[0], args[1], fileFilter);
  36.     }
  37.     
  38.     public void grep(String dirName, String expr, String filter) throws Exception {
  39.         
  40.         // 対象ファイルを取得
  41.         File dir = new File(dirName);
  42.         final String fltr = filter;
  43.         File[] files = dir.listFiles(
  44.                 new FilenameFilter(){
  45.                     public boolean accept(File dir, String name) {
  46.                         return name.matches(fltr);
  47.                     }
  48.                 });
  49.         
  50.         // 出力先ディレクトリ
  51.         File outDir = new File(dir.getAbsolutePath() + File.separator + "out");
  52.         if (!outDir.exists()) {
  53.             if (!outDir.mkdirs()) {
  54.                 throw new IllegalStateException("出力ディレクトリが作成できません");
  55.             }
  56.         }
  57.         
  58.         // スレッドで処理させる単位(バケツ)のサイズを決定
  59.         int buketSize = files.length / THREAD_SIZE;
  60.         buketSize++;
  61.         
  62.         int filesInBuketCnt = 0;
  63.         List<List<File>> buketBuket = new ArrayList<List<File>>();
  64.         List<File> fileBuket = null;
  65.         
  66.         for (File file : files) {
  67.             if (filesInBuketCnt == 0) {
  68.                 fileBuket = new ArrayList<File>();
  69.             }
  70.             fileBuket.add(file);
  71.             filesInBuketCnt++;
  72.             if (filesInBuketCnt > buketSize) {
  73.                 buketBuket.add(fileBuket);
  74.                 filesInBuketCnt = 0;
  75.                 fileBuket = null;
  76.             }
  77.         }
  78.         if (files !=null && filesInBuketCnt > 0) {
  79.             buketBuket.add(fileBuket);
  80.         }
  81.         
  82.         // スレッド数を登録
  83.         workingThreadCount = new AtomicInteger(buketBuket.size());
  84.  
  85.         // 非同期処理を行う
  86.         Executor executer = new Executor() {
  87.             public void execute(Runnable command) {
  88.                 new Thread(command).start();
  89.             }
  90.         };
  91.         
  92.         int threadId = 1;
  93.         for (List<File> lstFiles : buketBuket) {
  94.             executer.execute(
  95.                 new GrepCommand(threadId++,
  96.                         (File[])lstFiles.toArray(new File[0]),
  97.                         outDir, 
  98.                         expr));
  99.         }
  100.     }
  101.     
  102.     /**
  103.      * grep を実行する
  104.      */
  105.     public static class GrepCommand implements Runnable {
  106.         private int threadId;
  107.         private File[] files;
  108.         private File outDir;
  109.         private String expr;
  110.         
  111.         /**
  112.          * 対象ファイルから、正規表現に一致する行を、出力ディレクトリにスレッド識別用IDファイル名にて結果出力する
  113.          * @param threadId スレッド識別用ID
  114.          * @param files grep 対象ファイル
  115.          * @param outDir 結果出力ディレクトリ
  116.          * @param expr 正規表現(Java)
  117.          */
  118.         public GrepCommand(int threadId, File[] files, File outDir, String expr) {
  119.             if (outDir == null || !outDir.isDirectory()) {
  120.                 throw new IllegalArgumentException("不正な出力先");
  121.             }
  122.             this.threadId = threadId;
  123.             this.files = files;
  124.             this.outDir = outDir;
  125.             this.expr = expr;
  126.         }
  127.         
  128.         /* (non-Javadoc)
  129.          * @see java.lang.Runnable#run()
  130.          */
  131.         public void run() {
  132.             try {
  133.                 File outFile = new File(this.outDir.getAbsolutePath() 
  134.                         + File.separator
  135.                         + String.format("%03d.txt", this.threadId));
  136.                 BufferedWriter writer = new BufferedWriter(
  137.                         new OutputStreamWriter(new FileOutputStream(outFile)));
  138.                 
  139.                 for (File file : files) {
  140.                     System.out.printf("[%03d]処理中 ... %s\n", this.threadId, file.getName());
  141.                     BufferedReader reader = new BufferedReader(
  142.                             new InputStreamReader(new FileInputStream(file)));
  143.                     
  144.                     long lineno = 0;
  145.                     boolean isFirstMatch = false;
  146.                     String line = null;
  147.                     while((line = reader.readLine()) != null) {
  148.                         lineno++;
  149.                         if (line.matches(this.expr)) {
  150.                             if (!isFirstMatch) {
  151.                                 writer.write("\n--------------------------------\n");
  152.                                 writer.write(file.getAbsolutePath());
  153.                                 writer.write("\n--------------------------------\n");
  154.                                 isFirstMatch = true;
  155.                             }
  156.                             writer.write(line + "\n");
  157.                         }
  158.                     }
  159.                     reader.close();
  160.                 }
  161.                 
  162.                 writer.close();
  163.                 
  164.             } catch (Exception e) {
  165.                 System.out.println(e);
  166.                 System.exit(-1);
  167.             } finally {
  168.                 // 動いているスレッドが無くなったら、終了メッセージを出力
  169.                 int remain = workingThreadCount.addAndGet(-1);
  170.                 System.out.println("残りのスレッド数... " + remain);
  171.                 if ( remain <= 0) {
  172.                     System.out.println("終了しました。");
  173.                 }
  174.             }
  175.         }
  176.     }
  177. }
実行例11
  1. > java JGrep "C:\\work\\" ".*TEST.*" ".*\.txt$"
  2.  
  3. [006]処理中 ... WORK16.txt
  4. [005]処理中 ... WORK11.txt
  5. [007]処理中 ... WORK21.txt
  6. [001]処理中 ... TEST1028.txt
  7. [002]処理中 ... TEST2085.txt
  8. [002]処理中 ... TEST2089.txt
  9. [002]処理中 ... TEST2096.txt
  10. [001]処理中 ... TEST1135.txt
  11. [006]処理中 ... WORK17.txt
  12. [007]処理中 ... WORK23.txt
  13. [008]処理中 ... WORK25_M25.txt
  14. 残りのスレッド数... 7
  15. [006]処理中 ... WORK17.h.txt
  16. [001]処理中 ... TEST2076.txt
  17. [007]処理中 ... WORK24.txt
  18. [006]処理中 ... WORK19.txt
  19. [002]処理中 ... TEST2097Z.txt
  20. 残りのスレッド数... 6
  21. [007]処理中 ... WORK25#.txt
  22. 残りのスレッド数... 5
  23. [005]処理中 ... WORK11_M24.txt
  24. 残りのスレッド数... 4
  25. 残りのスレッド数... 3
  26. [008]処理中 ... WORK27.txt
  27. [005]処理中 ... WORK11_NEW.txt
  28. [008]処理中 ... コピー ～ TEST2085.txt
  29. [005]処理中 ... WORK11_W.txt
  30. [007]処理中 ... WORK25.20090630.txt
  31. 残りのスレッド数... 2
  32. [007]処理中 ... WORK25.txt
  33. [005]処理中 ... WORK11_WW.txt
  34. [005]処理中 ... WORK12.txt
  35. 残りのスレッド数... 1
  36. 残りのスレッド数... 0
  37. 終了しました。

非同期に grep を実施する例2

内容

  • 上記例を書き換え
  • ExecutorService を使って複数スレッドを管理
  • Runnable の代わりに、Callableを利用して、結果を受け取る
  • Future にて結果を表示

ExecutorService

  • 終了を管理するメソッド、および 1 つ以上の非同期タスクの進行状況を追跡する Future を生成できるメソッドを提供

Future

  • 非同期計算の結果を表す
  • 計算が完了したかどうかのチェック、完了までの待機、計算結果の取得などを行うためのメソッドを用意

Callable

  • 結果を返し、例外をスローすることがあるタスク
  • Runnable と似ていて、どちらもインスタンスが別のスレッドによって実行される可能性があるクラス用に設計
  • Runnable は結果を返さず、チェック例外をスローすることはできません


ソース

  1. import java.io.BufferedReader;
  2. import java.io.BufferedWriter;
  3. import java.io.File;
  4. import java.io.FileInputStream;
  5. import java.io.FileOutputStream;
  6. import java.io.FilenameFilter;
  7. import java.io.InputStreamReader;
  8. import java.io.OutputStreamWriter;
  9. import java.util.ArrayList;
  10. import java.util.List;
  11. import java.util.concurrent.Callable;
  12. import java.util.concurrent.ExecutorService;
  13. import java.util.concurrent.Executors;
  14. import java.util.concurrent.Future;
  15.  
  16. /**
  17.  * 
  18.  */
  19. public class JGrep {
  20.  
  21.     private static final int THREAD_SIZE = 10;
  22.     // スレッドプール
  23.     private ExecutorService threadPool;
  24.     
  25.     public static void main(String[] args) throws Exception {
  26.     
  27.         if (args.length <=2) {
  28.             System.out.println("Usage: java JGrep 対象ディレクトリ  正規表現  [ファイルフィルタ]");
  29.             System.exit(-1);
  30.         }
  31.  
  32.         String fileFilter = "";
  33.         if (args.length > 2) {
  34.             fileFilter = args[2];
  35.         }
  36.         (new JGrep()).grep(args[0], args[1], fileFilter);
  37.     }
  38.     
  39.     /**
  40.      * @param dirName
  41.      * @param expr
  42.      * @param filter
  43.      * @throws Exception
  44.      */
  45.     public void grep(String dirName, String expr, String filter) throws Exception {
  46.         
  47.         // 対象ファイルを取得
  48.         File dir = new File(dirName);
  49.         final String fltr = filter;
  50.         File[] files = dir.listFiles(
  51.                 new FilenameFilter(){
  52.                     public boolean accept(File dir, String name) {
  53.                         return name.matches(fltr);
  54.                     }
  55.                 });
  56.         
  57.         // 出力先ディレクトリ
  58.         File outDir = new File(dir.getAbsolutePath() + File.separator + "out");
  59.         if (!outDir.exists()) {
  60.             if (!outDir.mkdirs()) {
  61.                 throw new IllegalStateException("出力ディレクトリが作成できません");
  62.             }
  63.         }
  64.         
  65.         // スレッドで処理させる単位(バケツ)のサイズを決定
  66.         int buketSize = files.length / THREAD_SIZE;
  67.         buketSize++;
  68.         
  69.         int filesInBuketCnt = 0;
  70.         List<List<File>> buketBuket = new ArrayList<List<File>>();
  71.         List<File> fileBuket = null;
  72.         
  73.         for (File file : files) {
  74.             if (filesInBuketCnt == 0) {
  75.                 fileBuket = new ArrayList<File>();
  76.             }
  77.             fileBuket.add(file);
  78.             filesInBuketCnt++;
  79.             if (filesInBuketCnt > buketSize) {
  80.                 buketBuket.add(fileBuket);
  81.                 filesInBuketCnt = 0;
  82.                 fileBuket = null;
  83.             }
  84.         }
  85.         if (files !=null && filesInBuketCnt > 0) {
  86.             buketBuket.add(fileBuket);
  87.         }
  88.         
  89.         // Callable は、Runnabla と同じような役割を担うが、
  90.         // Runnable と異なり、結果を返し、例外をスローすることができる
  91.         List<Callable<Long>> tasks = new ArrayList<Callable<Long>>();
  92.         threadPool = Executors.newFixedThreadPool(THREAD_SIZE);
  93.  
  94.         int threadId = 1;
  95.         for (List<File> lstFiles : buketBuket) {
  96.             tasks.add(
  97.                 new GrepCommand(threadId++,
  98.                         (File[])lstFiles.toArray(new File[0]),
  99.                         outDir, 
  100.                         expr));
  101.         }
  102.         
  103.         // 指定されたタスクを実行し、すべて完了すると、ステータスと結果を含む Future のリストを返す
  104.         List<Future<Long>> results = threadPool.invokeAll(tasks);
  105.         for (Future<Long> f : results) {
  106.             System.out.printf("一致件数 %d 件\n", f.get().longValue());
  107.         }
  108.         System.out.println("終了しました。");
  109.         
  110.     }
  111.     
  112.     /**
  113.      * grep を実行する
  114.      */
  115.     public static class GrepCommand implements Callable<Long> {
  116.         private int threadId;
  117.         private File[] files;
  118.         private File outDir;
  119.         private String expr;
  120.         
  121.         /**
  122.          * 対象ファイルから、正規表現に一致する行を、出力ディレクトリにスレッド識別用IDファイル名にて結果出力する
  123.          * @param threadId スレッド識別用ID
  124.          * @param files grep 対象ファイル
  125.          * @param outDir 結果出力ディレクトリ
  126.          * @param expr 正規表現(Java)
  127.          */
  128.         public GrepCommand(int threadId, File[] files, File outDir, String expr) {
  129.             if (outDir == null || !outDir.isDirectory()) {
  130.                 throw new IllegalArgumentException("不正な出力先");
  131.             }
  132.             this.threadId = threadId;
  133.             this.files = files;
  134.             this.outDir = outDir;
  135.             this.expr = expr;
  136.         }
  137.  
  138.         /* (non-Javadoc)
  139.          * @see java.util.concurrent.Callable#call()
  140.          */
  141.         public Long call() throws Exception {
  142.             File outFile = new File(this.outDir.getAbsolutePath() 
  143.                     + File.separator
  144.                     + String.format("%03d.txt", this.threadId));
  145.             
  146.             BufferedWriter writer = new BufferedWriter(
  147.                     new OutputStreamWriter(new FileOutputStream(outFile)));
  148.             
  149.             long match = 0L;
  150.             for (File file : files) {
  151.                 System.out.printf("[%03d]処理中 ... %s\n", this.threadId, file.getName());
  152.                 BufferedReader reader = new BufferedReader(
  153.                         new InputStreamReader(new FileInputStream(file)));
  154.                 
  155.                 long lineno = 0;
  156.                 boolean isFirstMatch = false;
  157.                 String line = null;
  158.                 while((line = reader.readLine()) != null) {
  159.                     lineno++;
  160.                     if (line.matches(this.expr)) {
  161.                         if (!isFirstMatch) {
  162.                             writer.write("\n--------------------------------\n");
  163.                             writer.write(file.getAbsolutePath());
  164.                             writer.write("\n--------------------------------\n");
  165.                             isFirstMatch = true;
  166.                         }
  167.                         writer.write(line + "\n");
  168.                         match++;
  169.                     }
  170.                 }
  171.                 reader.close();
  172.             }
  173.             
  174.             writer.close();
  175.             
  176.             // 結果を返す
  177.             return new Long(match);
  178.         }
  179.     }
  180. }
「https://www.typea.info/tips_/index.php?title=Java_並行処理&oldid=6187」から取得

© 2006 矢木浩人