0

如果标题有点模糊，请原谅我。我将尝试更好地解释我要完成的工作。

有一个名为 parsebytes 的函数，它是我已实现的外部接口的一部分。它需要一个字节数组和一个长度。这个特定程序中的所有解析都在单个线程上运行，所以我想尽快从 parsebytes 中取出我的数据，以便它可以返回以离线获取更多数据。我在伪代码中的方法是这样的：创建一个外部运行的线程（ParserThreadClass）。每次调用 parsebytes 时，通过遍历所有字节并执行 byteQueue.add(bytes[i]) 将字节放入 ParserThreadClass 中的队列。这段代码被一个 synchronized(byteQueue) 包围，实际上，它应该释放 parsebytes 以返回并获取更多数据。

在这种情况下，我的 ParserThreadClass 也在运行。这是 run() 函数中的代码

while (!shutdown) //while the thread is still running
    {
        synchronized (byteQueue) 
        {
            bytes.addAll(byteQueue);  //an arraylist
            byteQueue.clear();
        }

        parseMessage();   //this will take the bytes arraylist and build an xml message.
    }

我在这里效率太低了吗？如果是这样，有人可以告诉我应该如何解决这个问题吗？

4

1 回答 1

2

这就是我之前尝试解决问题的方式。基本上你有一个生产者线程，就像你在这里一样，它读取文件并将项目放入队列中。然后你有一个工作线程从队列中读取并处理它们。代码如下，但它看起来与您正在做的基本相同。我发现这几乎没有让我加快速度，因为相对于磁盘读取而言，我需要每行执行的处理非常快。如果您必须进行的解析非常密集，或者块非常大，那么您可以看到这样做会加快一些速度。但如果它非常小，不要指望在性能改进方面看到太多，因为这个过程是 IO 绑定的。在这些情况下，您需要并行化磁盘访问，而这在单台机器上是无法做到的。

public static LinkedBlockingQueue<Pair<String, String>> mappings;
public static final Pair<String, String> end =
    new Pair<String, String>("END", "END");
public static AtomicBoolean done;
public static NpToEntityMapping mapping;
public static Set<String> attested_nps;
public static Set<Entity> possible_entities;

public static class ProducerThread implements Runnable {
    private File f;

    public ProducerThread(File f) {
        this.f = f;
    }

    public void run() {
        try {
            BufferedReader reader = new BufferedReader(new FileReader(f));
            String line;
            while ((line = reader.readLine()) != null) {
                String entities = reader.readLine();
                String np = line.trim();
                mappings.put(new Pair<String, String>(np, entities));
            }
            reader.close();
            for (int i=0; i<num_threads; i++) {
                mappings.put(end);
            }
        } catch (InterruptedException e) {
            System.out.println("Producer thread interrupted");
        } catch (IOException e) {
            System.out.println("Producer thread threw IOException");
        }
    }
}

public static class WorkerThread implements Runnable {
    private Dictionary dict;
    private EntityFactory factory;

    public WorkerThread(Dictionary dict, EntityFactory factory) {
        this.dict = dict;
        this.factory = factory;
    }

    public void run() {
        try {
            while (!done.get()) {
                Pair<String, String> np_ent = mappings.take();
                if (np_ent == end) {
                    done.set(false);
                    continue;
                }
                String entities = np_ent.getRight();
                String np = np_ent.getLeft().toLowerCase();
                if (attested_nps == null || attested_nps.contains(np)) {
                    int np_index = dict.getIndex(np);
                    HashSet<Entity> entity_set = new HashSet<Entity>();
                    for (String entity : entities.split(", ")) {
                        Entity e = factory.createEntity(entity.trim());
                        if (possible_entities != null) {
                            possible_entities.add(e);
                        }
                        entity_set.add(e);
                    }
                    mapping.put(np_index, entity_set);
                }
            }
        } catch (InterruptedException e) {
            System.out.println("Worker thread interrupted");
        }
    }
}

编辑：

这是启动生产者和工作线程的主线程的代码：

    Thread producer = new Thread(new ProducerThread(f), "Producer");
    producer.start();
    ArrayList<Thread> workers = new ArrayList<Thread>();
    for (int i=0; i<num_threads; i++) {
        workers.add(new Thread(new WorkerThread(dict, factory), "Worker"));
    }
    for (Thread t : workers) {
        t.start();
    }
    try {
        producer.join();
        for (Thread t : workers) {
            t.join();
        }
    } catch (InterruptedException e) {
        System.out.println("Main thread interrupted...");
    }

在生产者线程中完成的工作也应该在主线程中完成，从而无需在主代码中启动和加入另一个线程。不过，请务必在浏览文件之前启动工作线程，并在完成工作后加入它们。不过，我不确定这种方式与我在这里的方式之间的性能差异。

于 2012-11-19T14:49:24.140 回答