代码之家 › 专栏 › 技术社区 › Langali

从Java线程返回值

concurrency multithreading java

Langali · 技术社区 · 16 年前

我有如下Java线程:

   public class MyThread extends Thread {
        MyService service;
        String id;
        public MyThread(String id) {
            this.id = node;
        }
        public void run() {
            User user = service.getUser(id)
        }
    }

for(String id: ids) {
    MyThread thread = new MyThread(id);
    thread.start();
}

现在,我想收集每个线程的结果,并对数据库进行批插入,而不是每2秒插入300个数据库。

你知道我该怎么做吗?

9 回复 | 直到 16 年前

Community Mohan Dere 9 年前

如果要在执行数据库更新之前收集所有结果,可以使用 invokeAll 方法。如果您一次提交一个任务,那么这就需要进行簿记,比如 daveb

private static final ExecutorService workers = Executors.newCachedThreadPool();

...

Collection<Callable<User>> tasks = new ArrayList<Callable<User>>();
for (final String id : ids) {
  tasks.add(new Callable<User>()
  {

    public User call()
      throws Exception
    {
      return svc.getUser(id);
    }

  });
}
/* invokeAll blocks until all service requests complete, 
 * or a max of 10 seconds. */
List<Future<User>> results = workers.invokeAll(tasks, 10, TimeUnit.SECONDS);
for (Future<User> f : results) {
  User user = f.get();
  /* Add user to batch update. */
  ...
}
/* Commit batch. */
...

daveb 16 年前

规范的方法是使用 Callable ExecutorService . submit 到一个 执行者服务 返回a(typesafe) Future 从中你可以 get 结果。

class TaskAsCallable implements Callable<Result> {
    @Override
    public Result call() {
        return a new Result() // this is where the work is done.
    }
}

ExecutorService executor = Executors.newFixedThreadPool(300);
Future<Result> task = executor.submit(new TaskAsCallable());
Result result = task.get(); // this blocks until result is ready

invokeAll 返回一个 List 属于 Futures 得到 在每个人身上。

Bill K 16 年前

将结果存储到对象中。当它完成时,让它自己放入一个同步的集合中(脑海中浮现的是一个同步的队列)。

JDK中有很多工具可以帮助您实现这一点,但是一旦您开始将线程视为一个真正的对象,而不仅仅是一堆关于“run”方法的废话,就很容易了。一旦你开始用这种方式思考对象,编程就会变得更简单、更令人满意。

Sleiman Jneidi 10 年前

CompletableFuture . 假设我们有一个从数据库中获取id的类,为了简单起见,我们可以返回一个数字,如下所示,

static class GenerateNumber implements Supplier<Integer>{

    private final int number;

    GenerateNumber(int number){
        this.number = number;
    }
    @Override
    public Integer get() {
        try {
            TimeUnit.SECONDS.sleep(1);
        }catch (InterruptedException e){
            e.printStackTrace();
        }
        return this.number;
    }
}

现在,我们可以将结果添加到一个并发集合中,只要每个将来的结果都准备好了。

Collection<Integer> results = new ConcurrentLinkedQueue<>();
int tasks = 10;
CompletableFuture<?>[] allFutures = new CompletableFuture[tasks];
for (int i = 0; i < tasks; i++) {
     int temp = i;
     CompletableFuture<Integer> future = CompletableFuture.supplyAsync(()-> new GenerateNumber(temp).get(), executor);
     allFutures[i] = future.thenAccept(results::add);
 }

CompletableFuture.allOf(allFutures).thenAccept(c->{
   System.out.println(results); // do something with result
});

fastcodejava 16 年前

您需要将结果存储在类似singleton的文件中。必须正确同步。

编辑 Threads

rsp 16 年前

您可以创建一个队列或列表,将其传递给所创建的线程,线程将其结果添加到由执行批插入的使用者清空的列表中。

AndiDog 16 年前

最简单的方法是将一个对象传递给以后将包含结果的每个线程(每个线程一个对象)。主线程应该保留对每个结果对象的引用。连接所有线程后,可以使用结果。

Oso 16 年前

public class TopClass {
     List<User> users = new ArrayList<User>();
     void addUser(User user) {
         synchronized(users) {
             users.add(user);
         }
     }
     void store() throws SQLException {
        //storing code goes here
     }
     class MyThread extends Thread {
            MyService service;
            String id;
            public MyThread(String id) {
                this.id = node;
            }
            public void run() {
                User user = service.getUser(id)
                addUser(user);
            }
        }
}

jonescb Denis 16 年前

你可以使一个类扩展为可观察的。然后您的线程可以调用Observable类中的一个方法,该方法将通过调用Observable.notifyObservers(Object)来通知在该观察器中注册的任何类。

观察类将实现observator,并向可观察对象注册自己。然后,您将实现一个update(Observable,Object)方法,在调用Observable.notifyobservators(Object)时调用该方法。