Java:使用 JaxB 编码到 XML，如何正确地进行多线程

Question

我正在尝试获取一个很长的字符串文件，并根据给定的模式将其转换为 XML。我使用 jaxB 从该模式创建类。由于文件非常大，我创建了一个线程池来提高性能，但从那时起，它只处理文件的一行，并将其编码到 XML 文件，每个线程。

下面是我从文件中读取的家庭类(class)。每一行都是一个事务的记录，对于遇到的每个新用户，都会创建一个列表来存储所有该用户的事务，并将每个列表放入一个 HashMap 中。我将其设为 ConcurrentHashMap，因为多个线程将同时在映射上工作，这是正确的做法吗？

创建列表后，将为每个用户创建一个线程。每个线程都运行下面的 ProcessCommands 方法，并从 home 接收其用户的事务列表。

public class home{
  public static File XMLFile = new File("LogFile.xml");
  Map<String,List<String>> UserMap= new ConcurrentHashMap<String,List<String>>();
  String[] UserNames =  new String[5000];
    int numberOfUsers = 0;
    try{
        BufferedReader reader = new BufferedReader(new FileReader("test.txt"));
            String line;
            while ((line = reader.readLine()) != null)
            {
                parsed = line.split(",|\\s+");
                if(!parsed[2].equals("./testLOG")){
                    if(Utilities.checkUserExists(parsed[2], UserNames) == false){ //User does not already exist
                        System.out.println("New User: " + parsed[2]);
                        UserMap.put(parsed[2],new ArrayList<String>());         //Create list of transactions for new user
                        UserMap.get(parsed[2]).add(line);                       //Add First Item to new list
                        UserNames[numberOfUsers] = parsed[2];                   //Add new user
                        numberOfUsers++;
                    }
                    else{                                                           //User Already Existed
                        UserMap.get(parsed[2]).add(line);
                    }
                }
            }
            reader.close();
    } catch (IOException x) {
        System.err.println(x);
    }

    //get start time
    long startTime = new Date().getTime();
    tCount = numberOfUsers;
    ExecutorService threadPool = Executors.newFixedThreadPool(tCount);
    for(int i = 0; i < numberOfUsers; i++){
        System.out.println("Starting Thread " + i + " for user " + UserNames[i]);
        Runnable worker = new ProcessCommands(UserMap.get(UserNames[i]),UserNames[i], XMLfile);
        threadPool.execute(worker);
    }
    threadPool.shutdown();
    while(!threadPool.isTerminated()){

    }
    System.out.println("Finished all threads");

}

这是 ProcessCommands 类。该线程接收其用户的列表并创建一个编码器。据我了解，编码不是线程安全的，因此最好为每个线程创建一个，这是最好的方法吗？

当我创建编码器时，我知道每个来自(每个线程)都会想要访问创建的文件，从而导致冲突，我使用了同步，这是正确的吗？

当线程迭代其列表时，每行都调用特定的情况。有很多，所以为了清楚起见，我只是做了伪案例。每种情况都会调用下面的函数。

public class ProcessCommands implements Runnable{
private static final boolean DEBUG = false;
private List<String> list = null;
private String threadName;
private File XMLfile = null;
public Thread myThread;


public ProcessCommands(List<String> list, String threadName, File XMLfile){
    this.list = list;
    this.threadName = threadName;
    this.XMLfile = XMLfile;
}

public void run(){
    Date start = null;
    int transactionNumber = 0;
    String[] parsed = new String[8];
    String[] quoteParsed = null;
    String[] universalFormatCommand = new String[9];
    String userCommand = null;
    Connection connection = null;
    Statement stmt = null;
    Map<String, UserObject> usersMap = null;
    Map<String, Stack<BLO>> buyMap = null;
    Map<String, Stack<SLO>> sellMap = null;
    Map<String, QLO> stockCodeMap = null;
    Map<String, BTO> buyTriggerMap = null;
    Map<String, STO> sellTriggerMap = null;
    Map<String, USO> usersStocksMap = null;
    String SQL = null;
    int amountToAdd = 0;
    int tempDollars = 0;
    UserObject tempUO = null;
    BLO tempBLO = null;
    SLO tempSLO = null;
    Stack<BLO> tempStBLO = null;
    Stack<SLO> tempStSLO = null;
    BTO tempBTO = null;
    STO tempSTO = null;
    USO tempUSO = null;
    QLO tempQLO = null;
    String stockCode = null;
    String quoteResponse = null;
    int usersDollars = 0;
    int dollarAmountToBuy = 0;
    int dollarAmountToSell = 0;
    int numberOfSharesToBuy = 0;
    int numberOfSharesToSell = 0;
    int quoteStockInDollars = 0;
    int shares = 0;
    Iterator<String> itr = null;

    int transactionCount = list.size();
    System.out.println("Starting "+threadName+" - listSize = "+transactionCount);

    //UO dollars, reserved
    usersMap  = new HashMap<String, UserObject>(3);  //userName -> UO

    //USO shares
    usersStocksMap = new HashMap<String, USO>(); //userName+stockCode -> shares

    //BLO code, timestamp, dollarAmountToBuy, stockPriceInDollars
    buyMap = new HashMap<String, Stack<BLO>>();  //userName -> Stack<BLO>

    //SLO code, timestamp, dollarAmountToSell, stockPriceInDollars
    sellMap = new HashMap<String, Stack<SLO>>();  //userName -> Stack<SLO>

    //BTO code, timestamp, dollarAmountToBuy, stockPriceInDollars
    buyTriggerMap = new ConcurrentHashMap<String, BTO>();  //userName+stockCode -> BTO

    //STO code, timestamp, dollarAmountToBuy, stockPriceInDollars
    sellTriggerMap = new HashMap<String, STO>();  //userName+stockCode -> STO

    //QLO timestamp, stockPriceInDollars
    stockCodeMap = new HashMap<String, QLO>();  //stockCode -> QLO



    //create user object and initialize stacks
    usersMap.put(threadName, new UserObject(0, 0));
    buyMap.put(threadName, new Stack<BLO>());
    sellMap.put(threadName, new Stack<SLO>());
    try {
        //Marshaller marshaller = getMarshaller();
        synchronized (this){
            Marshaller marshaller = init.jc.createMarshaller();
            marshaller.setProperty(Marshaller.JAXB_FORMATTED_OUTPUT, true);
            marshaller.setProperty(Marshaller.JAXB_FRAGMENT, true);
            marshaller.marshal(LogServer.Root,XMLfile);
            marshaller.marshal(LogServer.Root,System.out);
        }
    } catch (JAXBException M) {
        M.printStackTrace();
    }

    Date timing = new Date();
    //universalFormatCommand = new String[8];
    parsed = new String[8];
    //iterate through workload file
    itr = this.list.iterator();
    while(itr.hasNext()){
        userCommand = (String) itr.next(); 
        itr.remove();
        parsed = userCommand.split(",|\\s+");
        transactionNumber = Integer.parseInt(parsed[0].replaceAll("\\[", "").replaceAll("\\]", ""));
        universalFormatCommand = Utilities.FormatCommand(parsed, parsed[0]);
        if(transactionNumber % 100 == 0){
            System.out.println(this.threadName + " - " +transactionNumber+ " - "+(new Date().getTime() - timing.getTime())/1000);
        }
        /*System.out.print("UserCommand " +transactionNumber + ": ");
        for(int i = 0;i<8;i++)System.out.print(universalFormatCommand[i]+ " ");
        System.out.print("\n");*/
        //switch for user command
        switch (parsed[1].toLowerCase()) {

        case "One"
            *Do Stuff"
            LogServer.create_Log(universalFormatCommand, transactionNumber, CommandType.ADD);
            break;
        case "Two"
            *Do Stuff"
            LogServer.create_Log(universalFormatCommand, transactionNumber, CommandType.ADD);
            break;
        }
     }
  }

函数 create_Log 有多种情况，和以前一样，为了清楚起见，我只留下了一种。案例“QUOTE”仅调用一个对象创建函数，但其​​他案例可以创建多个对象。 “log”类型是一个复杂的 XML 类型，它定义了所有其他对象类型，因此在每次调用 create_Log 时，我都会创建一个名为 Root 的日志类型。 JaxB 生成的“log”类包含一个创建对象列表的函数。声明:

Root.getUserCommandOrQuoteServerOrAccountTransaction().add(quote_QuoteType);

采用我创建的根元素，创建一个列表并将新创建的对象“quote_QuoteType”添加到该列表中。在添加线程之前，此方法成功创建了一个包含我想要的尽可能多的对象的列表，然后对它们进行编码。所以我非常肯定“LogServer”类中的部分不是问题。这与上面 ProcessCommands 类中的编码和同步有关。

public class LogServer{
    public static log Root = new log();

    public static QuoteServerType Log_Quote(String[] input, int TransactionNumber){
    ObjectFactory factory = new ObjectFactory();
    QuoteServerType quoteCall = factory.createQuoteServerType();

    **Populate the QuoteServerType object called quoteCall**

    return quoteCall;
    }

    public static void create_Log(String[] input, int TransactionNumber, CommandType Command){
    System.out.print("TRANSACTION "+TransactionNumber + " is " + Command + ": ");
    for(int i = 0; i<input.length;i++) System.out.print(input[i] + " ");
    System.out.print("\n");
    switch(input[1]){
    case "QUOTE":
        System.out.print("QUOTE CASE");
        QuoteServerType quote_QuoteType = Log_Quote(input,TransactionNumber);
        Root.getUserCommandOrQuoteServerOrAccountTransaction().add(quote_QuoteType);
        break;
        }
      }

Answer 1

所以你写了很多代码，但是你尝试过它是否真的有效吗？快速浏览后我对此表示怀疑。您应该逐个部分地测试您的代码逻辑，而不是一直测试到最后。看来你只是盯着Java。我建议首先在简单的单线程应用程序上进行练习。抱歉，如果我听起来很刺耳，但我也会尽力表现出建设性:

1. 按照约定，类名称以大写字母开头，变量以小写字母开头，您可以采用其他方式。
2. 您应该在您的主(Home)类中创建一个方法，而不是将所有代码放在静态 block 中。
3. 您正在将整个文件读取到内存中，而不是逐行处理它。 Home初始化后，文学文件的全部内容将位于UserMap变量下。如果文件非常大，您将耗尽堆内存。如果您假设文件很大，则无法做到这一点，并且您必须重新设计您的应用程序以将部分结果存储在某处。如果你的文件小于内存，你可以这样保留它(但你说它很大)。
4. 不需要 UserNames，UserMap.containsKey 即可完成工作
5. 你的线程池大小应该在你的核心范围内，而不是用户数量，因为你会得到线程垃圾(如果你的代码中有阻塞操作，如果不将其保留为处理器数量，则使 tCount = 2*processors)。一旦一个 ProcessCommand 完成，执行器将启动另一个 ProcessCommand，直到您完成所有任务，并且您将有效地使用所有处理器核心。
6. 不要 while(!threadPool.isTermminate())，这一行将完全消耗一个处理器，因为它将不断检查，而是调用awaitTermination
7. 您的 ProcessCommand 具有 View 映射变量，这些变量只有一个条目，因为正如您所说，每个条目都会处理来自一个用户的数据。
8. 同步(this)是进程将不起作用，因为每个线程将在不同的对象(进程的不同实例)上同步。
9. 我相信创建编码器是线程安全的(检查一下)，因此根本不需要同步
10. 您在对交易列表进行实际处理之前保存日志(无论是什么)
11. 编码将用 LogServer.Root 的当前状态覆盖文件内容。如果它在您的 proccsCommand 之间共享(看起来是这样)，那么将其保存在每个线程中有何意义。完成后就做。
12. 您不需要 itr.remove();
13. 日志类(对于 ROOT 变量!!!)需要是线程安全的，因为所有线程都会调用其上的操作(因此日志类内的列表必须是并发列表等)。
14. 诸如此类......

我会推荐，

1. 从实际有效的简单单线程版本开始。
2. 逐行进行处理(将每个用户的结果存储在不同的文件中，您可以为最近使用的用户的事务进行缓存，这样就不必一直写入磁盘(请参阅 Guava 缓存)
3. 将每个用户事务多线程处理到您的用户日志对象(同样，如果数量很多，您必须将它们保存到磁盘而不是全部保存在内存中)。
4. 编写代码，将不同用户的日志组合起来创建一个(您可能再次希望以多线程的方式进行操作)，尽管这主要是 IO 操作，因此不会带来太多好处，而且做起来也比较棘手。

祝你好运 覆盖续