执行器框架异常行为

Question

您好，我正在使用 Executors 并行加载一些数据。我的应用程序正在从具有父子关系的数据库中获取一些数据，例如：

parent 1 -> [child11, child12,..., child1N]
parent 2 -> [child21, childy22,..., child2N]
.....
parent N -> [childN1, childyN2,..., childNN]

现在我想要并行处理。我现在正在做的是加载所有数据一次从数据库中设置一个父子并调用执行程序服务来映射关系中的那些并存储在我的数据结构中。

现在我的代码是：

父子关系如下：

public class Post implements Serializable {

    private static final long serialVersionUID = 1L;

    private Integer postId;
    private String  postText;
    private String  postType;
    private Integer menuItemId;
    private boolean parentPost;
    private Integer parentPostId;
    // Contains all the Child of this Post
    private List<Post> answers = new ArrayList<Post>();
    ....
    //getters and setters
}

现在我有一个包装器用于同步Postclass

public class PostList {

    private List<Post> postList;

    public PostList() {
        super();
        this.postList = new ArrayList<Post>();
    }

    public List<Post> getPostList() {
        return postList;
    }

    public synchronized boolean add(Post post) {
        return postList.add(post);
    }

    public synchronized boolean addAnswer(Post answer) {
        for(Post post : postList)
        {
            if(post.getPostId() == answer.getParentPostId())
            {
                post.getAnswers().add(answer);
                break;
            }
        }
        return true;
    }
}

现在我从数据库加载的代码是：

/* This is called to load each parent-child set at a time, when the 
first set is fetched from DB then call to executor to store those in  
internal data structure. */

List<Post> posts = null;
PostList postList = null;
Integer args[] ={menuItemId};
// Fetch all Posts which are in parent child relation
posts = getDataFromDB(...)
if(posts != null && posts.size() >0)
{
    postList = new PostList();
    ExecutorService executor = Executors.newFixedThreadPool(10);
    for(Post post : posts)
    {
         executor.execute(new PostProcessor(post, postList));
    }
    logger.debug("Starting executor shutdown...");
    executor.shutdown();
    while (!executor.isTerminated()) {
        try {
            executor.awaitTermination(1000, TimeUnit.MILLISECONDS);
        } catch (InterruptedException ex) {
            logger.error("Interrupted executor >>", ex.getMessage(), ex);
        }
    }
    logger.debug("All post loading done ...");
    logger.debug("PostList >> " + postList);
    if(postList.getPostList() != null)
        return postList.getPostList();
}

在Post处理器中我有

public class PostProcessor implements Runnable {

    private Post post;
    private PostList postList;


    public PostProcessor(Post post, PostList postList) {
        super();
        this.post = post;
        this.postList = postList;
    }

    @Override
    public void run() {
        // TODO Auto-generated method stub
        Post answer = null;
        try
        {
            // if Post is parent / is a question
            if ("Q".equalsIgnoreCase(post.getPostType())) 
            {
                // do some operation
                postList.add(post);
            }
            // Post is an Answer, so add the answer to proper Question
            else {
                answer = post;
                postList.addAnswer(answer);
            }
            Thread.sleep(1000);
        }
        catch(Throwable throwable)
        {
            logger.error(throwable.getMessage(),throwable);
        }
    }
}

但它表现异常，有时它加载所有问题 post 但不是所有答案，有时它根本不加载父 post。请帮助我哪里做错了。

Answer 1

如果addAnswer 失败那么它应该return false 或抛出异常；这表明相应的问题尚未加载或不存在。两个选项：

首先处理所有问题，如果答案与问题不匹配则抛出异常。
当您查询数据库时，获取问题的计数并在每次处理问题时将其递减（在将问题添加到 [=29= 之后递减 ] 列表，否则你可能会得到一个 question_count == 0 但还没有问题被添加到列表中）；如果答案与问题不匹配并且 question_count > 0 则将答案放回队列，否则抛出异常。

与其说是正确性，不如说是效率问题，我建议您取消 synchronized 方法并改用 java.util.concurrent 中的线程安全数据结构 - 这将减少锁争用。这看起来像

public class PostList { private AtomicInteger questionCount; private ConcurrentLinkedQueue<Post> questions; private ConcurrentHashMap<String, ConcurrentLinkedQueue<Post>> answers; public boolean addQuestion(Post post) { questions.offer(post); if(answers.putIfAbsent(post.getPostId(), new ConcurrentLinkedQueue<>()) != null) { questionCount.decrementAndGet(); return true; } else throw new IllegalArgumentException("duplicate question id"); } public boolean addAnswer(Post answer) { ConcurrentLinkedQueue<Post> queue = answers.get(answer.getParentPostId()); if(queue != null) { queue.offer(answer); return true; } else if(questionCount.get() > 0) { return false; } else { throw new IllegalArgumentException("answer has no question"); } } }

执行器框架异常行为

Executor framework abnormal behaviour

java

concurrency

multithreading

executorservice

java.util.concurrent