Как создавать документы PDF из rpt в многопоточном подходе?

Ответ 1

Это очень простой код. A ThreadPoolExecutor с потоками фиксированного размера в пуле является основой.

Некоторые соображения:

• Размер пула потоков должен быть равен или меньше размера пула подключений к БД. И это должно быть оптимальное число, которое разумно для параллельных двигателей.
• Основной поток должен ждать достаточно времени, прежде чем убить все потоки. Я поставил 1 час как время ожидания, но это всего лишь пример.
• Вам нужно иметь правильную обработку Исключения.
• Из документа API я увидел методы stopAll и shutdown из класса Engine. Поэтому я призываю это, как только наша работа будет завершена. Это снова, просто пример.

Надеюсь, что это поможет.

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.sql.Connection;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class RunEngine {
    public static void main(String[] args) throws Exception {
        final String rpt = "/tmp/rpt/input/rpt-1.rpt";
        final String sFilePath = "/tmp/rpt/output/";
        final String sFileName = "pdfreport";
        final Object[] data = new Object[10];

        ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(10);
        for (int i = 0; i < data.length; i++) {
            PDFExporterRunnable runnable = new PDFExporterRunnable(rpt, data[i], sFilePath, sFileName, i);
            executor.execute(runnable);
        }
        executor.shutdown();
        executor.awaitTermination(1L, TimeUnit.HOURS);
        Engine.stopAll();
        Engine.shutdown();
    }
    private static class PDFExporterRunnable implements Runnable {
        private final String rpt;
        private final Object data;
        private final String sFilePath;
        private final String sFileName;
        private final int runIndex;


        public PDFExporterRunnable(String rpt, Object data, String sFilePath,
                String sFileName, int runIndex) {
            this.rpt = rpt;
            this.data = data;
            this.sFilePath = sFilePath;
            this.sFileName = sFileName;
            this.runIndex = runIndex;
        }

        @Override
        public void run() {
            // Loops
            Engine eng = new Engine(Engine.EXPORT_PDF);
            eng.setReportFile(rpt); // rpt is the report name
            Connection cn = null;

            /*
             * DB connection related code. Check and use.
             */
            //if (cn.isClosed() || cn == null) {
                //cn = ds.getConnection();
            //}
            eng.setConnection(cn);
            System.out.println(" After set connection");

            eng.setPrompt(data, 0);
            ReportProperties repprop = eng.getReportProperties();
            repprop.setPaperOrient(ReportProperties.DEFAULT_PAPER_ORIENTATION,
                    ReportProperties.PAPER_FANFOLD_US);
            eng.execute();
            System.out.println(" After excecute");
            FileOutputStream fos = null;
            try {
                String FileName = sFileName + "_" + runIndex;
                File file = new File(sFilePath + FileName + ".pdf");
                if (!file.getParentFile().exists()) {
                    file.getParentFile().mkdirs();
                }
                if (!file.exists()) {
                    file.createNewFile();
                }
                fos = new FileOutputStream(file);
                for (int k = 1; k <= eng.getPageCount(); k++) {
                    fos.write(eng.getPageData(k));
                }
                fos.flush();
                fos.close();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (fos != null) {
                    try {
                        fos.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                    fos = null;
                }
            }
        }
    }
    /*
     * Dummy classes to avoid compilation errors.
     */
    private static class ReportProperties {
        public static final String PAPER_FANFOLD_US = null;
        public static final String DEFAULT_PAPER_ORIENTATION = null;
        public void setPaperOrient(String defaultPaperOrientation, String paperFanfoldUs) {
        }
    }

    private static class Engine {
        public static final int EXPORT_PDF = 1;
        public Engine(int exportType) {
        }
        public static void shutdown() {
        }
        public static void stopAll() {
        }
        public void setPrompt(Object singleData, int i) {
        }
        public byte[] getPageData(int k) {
            return null;
        }
        public int getPageCount() {
            return 0;
        }
        public void execute() {
        }
        public ReportProperties getReportProperties() {
            return null;
        }
        public void setConnection(Connection cn) {
        }
        public void setReportFile(String reportFile) {
        }
    }
}

Ответ 2

Я предлагаю этот "ответ" как возможное быстрое и грязное решение, чтобы вы начали работу по распараллеливанию.

Так или иначе вы собираетесь создать ферму рендеринга. Я не думаю, что есть тривиальный способ сделать это в java; Я бы хотел, чтобы кто-то опубликовал ответ, который показывает, как распараллелить ваш пример всего несколькими строками кода. Но пока это не произойдет, это, надеюсь, поможет вам добиться определенного прогресса.

У вас будет ограниченное масштабирование в одном экземпляре JVM. Но... давайте посмотрим, как далеко вы справитесь с этим, и посмотрите, помогает ли он достаточно.

Задача проектирования №1: перезапуск.

Вам, вероятно, понадобится место для сохранения статуса для каждого из ваших отчетов, например. "единицы работы".

Вы хотите, чтобы это произошло, если вам нужно перезапустить все (возможно, сбой вашего сервера), и вы не хотите повторно запускать все отчеты до сих пор.

Множество способов сделать это; базы данных, проверьте, существует ли в папке отчета "завершенный" файл (недостаточно для существования *.pdf, поскольку это может быть неполным... для xyz_200.pdf вы можете создать пустой файл xyz_200.done или xyz_200.err, чтобы помочь с повторным запуском каких-либо проблемных детей... и к тому времени, когда вы кодируете эту логику управления файлами/проверки/инициализации, кажется, что было проще добавить столбец в вашу базу данных, в котором содержится список работа будет выполнена).

Рассмотрение конструкции № 2: максимальная пропускная способность (исключая перегрузку).

Вы не хотите насыщать свою систему и запускать тысячу отчетов параллельно. Может быть, 10.
Может быть, 100.
Вероятно, не 5000.
Вам нужно будет провести какое-то исследование размеров и посмотреть, что вы получаете от 80 до 90% использования системы.

Рассмотрение дизайна №3: ​​масштабирование на нескольких серверах

Слишком сложный, выходящий за рамки ответа на Stack Exchange. Вам придется разворачивать JVM на нескольких системах, которые работают как рабочие внизу, а менеджер отчетов, который может вытаскивать рабочие элементы из общей структуры "очереди", снова, вероятно, здесь лучше работает таблица базы данных, чем что-то делать (или сетевой канал).

Пример кода

Внимание: ни один из этих кодов не проверен, он почти наверняка имеет обилие опечаток, логические ошибки и плохой дизайн. Используйте на свой страх и риск.

Так или иначе... Я хочу дать вам основную идею рудиментарного задания. Замените пример "//Loops" в вопросе кодом следующим образом:

основной цикл (пример исходного кода)

Это более или менее то, что сделал ваш примерный код, модифицированный, чтобы проталкивать большую часть работы в ReportWorker (новый класс, см. ниже). Многие вещи, кажется, упакованы в ваш исходный пример вопроса "//Loop", поэтому я не пытаюсь перепроектировать это.

fwiw, мне было непонятно, откуда берутся "rpt" и "data [i]", поэтому я взломал некоторые тестовые данные.

public class Main {

   public static boolean complete( String data ) {
      return false; // for testing nothing is complete.
   }

    public static void main(String args[] ) {

    String data[] = new String[] { 
         "A",
         "B",
         "C",
         "D",
         "E" };
    String rpt = "xyz";

    // Loop
    ReportManager reportMgr = new ReportManager();  // a new helper class (see below), it assigns/monitors work.
    long startTime = System.currentTimeMillis();
    for( int i = 0; i < data.length; ++i ) {
       // complete is something you should write that knows if a report "unit of  work"
       // finished successfully.
       if( !complete( data[i] ) ) {
          reportMgr.assignWork(  rpt, data[i] ); // so... where did values for your "rpt" variable come from?
       }
    }
    reportMgr.waitForWorkToFinish(); // out of new work to assign, let wait until everything in-flight complete.
    long endTime = System.currentTimeMillis();
    System.out.println("Done.  Elapsed time = " + (endTime - startTime)/1000 +" seconds.");

   }

}

ReportManager

Этот класс не является потокобезопасным, просто используйте свой исходный цикл для вызова assignWork() до тех пор, пока не выйдете из отчетов для назначения, затем продолжайте называть его до тех пор, пока все работы не будут выполнены, например. waitForWorkToFinish(), как показано выше. (fwiw, я не думаю, что вы могли бы сказать, что любой из классов здесь особенно безопасен в потоке).

public class ReportManager {

   public int polling_delay = 500; // wait 0.5 seconds for testing.
   //public int polling_delay = 60 * 1000; // wait 1 minute.
   // not high throughput millions of reports / second, we'll run at a slower tempo.
   public int nWorkers = 3; // just 3 for testing.
   public int assignedCnt = 0;
   public ReportWorker workers[];

   public ReportManager() {
      // initialize our manager.
      workers = new ReportWorker[ nWorkers ];
      for( int i = 0; i < nWorkers; ++i ) {
         workers[i] = new ReportWorker( i );
         System.out.println("Created worker #"+i);
      }
   }

   private ReportWorker handleWorkerError( int i  ) {
      // something went wrong, update our "report" status as one of the reports failed.
      System.out.println("handlerWokerError(): failure in "+workers[i]+", resetting worker.");
      workers[i].teardown();
      workers[i] = new ReportWorker( i ); // just replace everything.
      return workers[i]; // the new worker will, incidentally, be avaialble.
   }

   private ReportWorker handleWorkerComplete( int i ) {
      // this unit of work was completed, update our "report" status tracker as success.
      System.out.println("handleWorkerComplete(): success in "+workers[i]+", resetting worker.");
      workers[i].teardown();
      workers[i] = new ReportWorker( i ); // just replace everything.
      return workers[i]; // the new worker will, incidentally, be avaialble.
   }

   private int activeWorkerCount() {
      int activeCnt = 0;
      for( int i = 0; i < nWorkers; ++i ) {
         ReportWorker worker = workers[i];
         System.out.println("activeWorkerCount() i="+i+", checking worker="+worker);
         if( worker.hasError() ) {
            worker = handleWorkerError( i );
         }
         if( worker.isComplete() ) {
            worker = handleWorkerComplete( i );
         }
         if( worker.isInitialized() || worker.isRunning() ) {
            ++activeCnt;
         }
      }
      System.out.println("activeWorkerCount() activeCnt="+activeCnt);
      return activeCnt;
   }

   private ReportWorker getAvailableWorker() {
      // check each worker to see if anybody recently completed...
      // This (rather lazily) creates completely new ReportWorker instances.
      // You might want to try pooling (salvaging and reinitializing them)
      // to see if that helps your performance.

      System.out.println("\n-----");
      ReportWorker firstAvailable = null;
      for( int i = 0; i < nWorkers; ++i ) {
         ReportWorker worker = workers[i];
         System.out.println("getAvailableWorker(): i="+i+" worker="+worker);
         if( worker.hasError() ) {
            worker = handleWorkerError( i );
         }
         if( worker.isComplete() ) {
            worker = handleWorkerComplete( i );
         }
         if( worker.isAvailable() && firstAvailable==null ) {
            System.out.println("Apparently worker "+worker+" is 'available'");
            firstAvailable  = worker;
            System.out.println("getAvailableWorker(): i="+i+" now firstAvailable = "+firstAvailable);
         }
      }
      return firstAvailable;  // May (or may not) be null.
   }

   public void assignWork(  String rpt, String data ) {
      ReportWorker worker = getAvailableWorker();
      while( worker == null ) {
         System.out.println("assignWork: No workers available, sleeping for "+polling_delay);
         try { Thread.sleep( polling_delay ); }
         catch( InterruptedException e ) { System.out.println("assignWork: sleep interrupted, ignoring exception "+e); }
         // any workers avaialble now?
         worker = getAvailableWorker();
      }
      ++assignedCnt;
      worker.initialize( rpt, data ); // or whatever else you need.
      System.out.println("assignment #"+assignedCnt+" given to "+worker);
      Thread t = new Thread( worker );
      t.start( ); // that is pretty much it, let it go.
   }

   public void waitForWorkToFinish() {
      int active = activeWorkerCount();
      while( active >= 1 ) {
         System.out.println("waitForWorkToFinish(): #active workers="+active+", waiting...");
         // wait a minute....
         try { Thread.sleep( polling_delay ); }
         catch( InterruptedException e ) { System.out.println("assignWork: sleep interrupted, ignoring exception "+e); }
         active = activeWorkerCount();
      }
   }
}

ReportWorker

public class ReportWorker implements Runnable {
      int test_delay = 10*1000; //sleep for 10 seconds.
      // (actual code would be generating PDF output)

      public enum StatusCodes { UNINITIALIZED,
          INITIALIZED,
          RUNNING,
          COMPLETE,
          ERROR };


      int id = -1;
      StatusCodes status = StatusCodes.UNINITIALIZED;
      boolean initialized = false;
      public String rpt = "";
      public String data = "";
      //Engine eng;
      //PDFExportThread pdfExporter;
      //DataSource_type cn;

      public boolean isInitialized() { return initialized; }
      public boolean isAvailable()   { return status == StatusCodes.UNINITIALIZED; }
      public boolean isRunning()     { return status == StatusCodes.RUNNING; }
      public boolean isComplete()    { return status == StatusCodes.COMPLETE; }
      public boolean hasError()      { return status == StatusCodes.ERROR; }


      public ReportWorker( int id ) {
          this.id = id;
      }

      public String toString( ) {
         return "ReportWorker."+id+"("+status+")/"+rpt+"/"+data;
      }

      // the example code doesn't make clear if there is a relationship between rpt & data[i].
      public void initialize( String rpt, String data /* data[i] in original code */  ) {
         try {
            this.rpt = rpt;
            this.data = data;
            /* uncomment this part where you have the various classes availble.
             * I have it commented out for testing.
            cn = ds.getConnection();   
            Engine eng = new Engine(Engine.EXPORT_PDF);
            eng.setReportFile(rpt); //rpt is the report name
            eng.setConnection(cn);
            eng.setPrompt(data, 0);
            ReportProperties repprop = eng.getReportProperties();
            repprop.setPaperOrient(ReportProperties.DEFAULT_PAPER_ORIENTATION, ReportProperties.PAPER_FANFOLD_US);
            */
            status = StatusCodes.INITIALIZED;
            initialized = true; // want this true even if we're running.
         } catch( Exception e ) {
            status = StatusCodes.ERROR;
            throw new RuntimeException("initialze(rpt="+rpt+", data="+data+")", e);
         }
      }

      public void run() {
         status = StatusCodes.RUNNING;
         System.out.println("run().BEGIN: "+this);
         try {
            // delay for testing.
            try { Thread.sleep( test_delay ); }
            catch( InterruptedException e ) { System.out.println(this+".run(): test interrupted, ignoring "+e); }
            /* uncomment this part where you have the various classes availble.
             * I have it commented out for testing.
            eng.execute();
            PDFExportThread pdfExporter = new PDFExportThread(eng, sFileName, sFilePath);
            pdfExporter.execute();
            */
            status = StatusCodes.COMPLETE;
            System.out.println("run().END: "+this);
         } catch( Exception e ) {
            System.out.println("run().ERROR: "+this);
            status = StatusCodes.ERROR;
            throw new RuntimeException("run(rpt="+rpt+", data="+data+")", e);
         }
      }

      public void teardown() {
         if( ! isInitialized() || isRunning() ) {
            System.out.println("Warning: ReportWorker.teardown() called but I am uninitailzied or running.");
            // should never happen, fatal enough to throw an exception?
         }

         /* commented out for testing.
           try { cn.close(); } 
           catch( Exception e ) { System.out.println("Warning: ReportWorker.teardown() ignoring error on connection close: "+e); }
           cn = null;
         */
         // any need to close things on eng?
         // any need to close things on pdfExporter?
      }
}