blocks|key|1322120|text|规范的方法是使用Callable和ExecutorService。将Callable设置为ExecutorService会返回一个(类型安全的)+Future，您可以从中get结果。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1322121|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|code-block|syntax|javascript|1322122|在您的示例中，您可能希望使用返回Futures+List的invokeAll，或者在向executor添加任务时自己创建该列表。要收集结果，只需对每个结果调用get。|1322123|entityMap^0|8|8|H|F|Y|8|19|F|21|6|2D|3|0|0|G|7|O|4|T|9|27|3|0^^$0|@$1|2|3|4|5|6|7|O|8|@$9|P|A|Q|B|C]|$9|R|A|S|B|C]|$9|T|A|U|B|C]|$9|V|A|W|B|C]|$9|X|A|Y|B|C]|$9|Z|A|10|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|11|8|@]|D|@]|E|$I|J]]|$1|K|3|L|5|6|7|12|8|@$9|13|A|14|B|C]|$9|15|A|16|B|C]|$9|17|A|18|B|C]|$9|19|A|1A|B|C]]|D|@]|E|$]]|$1|M|3|-4|5|6|7|1B|8|@]|D|@]|E|$]]]|N|$]]

The canonical approach is to use a <code>Callable</code> and an <code>ExecutorService</code>. <code>submit</code>ting a <code>Callable</code> to an <code>ExecutorService</code> returns a (typesafe) <code>Future</code> from which you can <code>get</code> the result.

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

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

In your case, you probably want to use <code>invokeAll</code> which returns a <code>List</code> of <code>Futures</code>, or create that list yourself as you add tasks to the executor. To collect results, simply call <code>get</code> on each one.

blocks|key|1322134|text|您可以创建一个队列或列表，将其传递给您创建的线程，线程将其结果添加到列表中，该列表由执行批量插入的消费者清空。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1322135|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|D|8|@]|9|@]|A|$]]|$1|B|3|-4|5|6|7|E|8|@]|9|@]|A|$]]]|C|$]]

You could create a queue or list which you pass to the threads you create, the threads add their result to the list which gets emptied by a consumer which performs the batch insert.

blocks|key|1322166|text|最简单的方法是将一个对象传递给每个线程(每个线程一个对象)，该线程稍后将包含结果。主线程应该保留对每个结果对象的引用。当所有线程都联接后，就可以使用结果了。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1322167|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|D|8|@]|9|@]|A|$]]|$1|B|3|-4|5|6|7|E|8|@]|9|@]|A|$]]]|C|$]]

The simplest approach is to pass an object to each thread (one object per thread) that will contain the result later. The main thread should keep a reference to each result object. When all threads are joined, you can use the results.

blocks|key|1107042|text|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);
++++++++++++}
++++++++}
}|type|code-block|depth|inlineStyleRanges|entityRanges|data|syntax|javascript|1107043|unstyled|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|G|8|@]|9|@]|A|$B|C]]|$1|D|3|-4|5|E|7|H|8|@]|9|@]|A|$]]]|F|$]]

<pre><code>public class TopClass {
 List&lt;User&gt; users = new ArrayList&lt;User&gt;();
 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);
 }
 }
}
</code></pre>

blocks|key|1107078|text|如果希望在进行数据库更新之前收集所有结果，可以使用invokeAll方法。这就像daveb建议的那样，如果你一次提交一个任务，就需要记账。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1107079|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.+*/
...|code-block|syntax|javascript|1107080|entityMap|0|LINK|mutability|MUTABLE|url|http://java.sun.com/javase/6/docs/api/java/util/concurrent/ExecutorService.html#invokeAll%2528java.util.Collection%2529|1|https://stackoverflow.com/questions/2314402/return-values-from-java-threads/2314418#2314418^0|P|9|P|9|0|14|5|1|0|0^^$0|@$1|2|3|4|5|6|7|U|8|@$9|V|A|W|B|C]]|D|@$9|X|A|Y|1|Z]|$9|10|A|11|1|12]]|E|$]]|$1|F|3|G|5|H|7|13|8|@]|D|@]|E|$I|J]]|$1|K|3|-4|5|6|7|14|8|@]|D|@]|E|$]]]|L|$M|$5|N|O|P|E|$Q|R]]|S|$5|N|O|P|E|$Q|T]]]]

If you want to collect all of the results before doing the database update, you can use the <a href="http://java.sun.com/javase/6/docs/api/java/util/concurrent/ExecutorService.html#invokeAll%28java.util.Collection%29" rel="noreferrer"><code>invokeAll</code></a> method. This takes care of the bookkeeping that would be required if you submit tasks one at a time, like <a href="https://stackoverflow.com/questions/2314402/return-values-from-java-threads/2314418#2314418">daveb</a> suggests.

<pre><code>private static final ExecutorService workers = Executors.newCachedThreadPool();

...

Collection&lt;Callable&lt;User&gt;&gt; tasks = new ArrayList&lt;Callable&lt;User&gt;&gt;();
for (final String id : ids) {
 tasks.add(new Callable&lt;User&gt;()
 {

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

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

blocks|key|1322263|text|你可以创建一个扩展Observable的类。然后，您的线程可以调用Observable类中的方法，该方法将通过调用Observable.notifyObservers(Object)通知在该观察器中注册的任何类。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1322264|观察类将实现观察者，并将自己注册到观察者。然后，您将实现一个更新(Observable，Object)方法，该方法在调用Observerable.notifyObservers(Object)时被调用。|1322265|entityMap^0|0|0^^$0|@$1|2|3|4|5|6|7|F|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|G|8|@]|9|@]|A|$]]|$1|D|3|-4|5|6|7|H|8|@]|9|@]|A|$]]]|E|$]]

You could make a class which extends Observable. Then your thread can call a method in the Observable class which would notify any classes that registered in that observer by calling Observable.notifyObservers(Object).

The observing class would implement Observer, and register itself with the Observable. You would then implement an update(Observable, Object) method that gets called when Observerable.notifyObservers(Object) is called.

blocks|key|1107191|text|将结果存储在对象中。当它完成时，让它把自己放到一个同步的集合中(想到的是一个同步队列)。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1107192|当您希望收集要提交的结果时，从队列中抓取所有内容，并从对象中读取结果。您甚至可以让每个对象知道如何将它自己的结果"post“到数据库中，这样就可以提交不同的类，并使用完全相同的小而优雅的循环来处理所有类。|1107193|JDK中有很多工具可以帮助你做到这一点，但是一旦你开始把你的线程看作一个真正的对象，而不只是一堆关于"run“方法的废话，事情就变得很简单了。一旦你开始以这种方式思考对象，编程就会变得更加简单和令人满意。|1107194|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|H|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|I|8|@]|9|@]|A|$]]|$1|D|3|E|5|6|7|J|8|@]|9|@]|A|$]]|$1|F|3|-4|5|6|7|K|8|@]|9|@]|A|$]]]|G|$]]

Store your result in your object. When it completes, have it drop itself into a synchronized collection (a synchronized queue comes to mind).

When you wish to collect your results to submit, grab everything from the queue and read your results from the objects. You might even have each object know how to "post" it's own results to the database, this way different classes can be submitted and all handled with the exact same tiny, elegant loop.

There are lots of tools in the JDK to help with this, but it is really easy once you start thinking of your thread as a true object and not just a bunch of crap around a "run" method. Once you start thinking of objects this way programming becomes much simpler and more satisfying.

blocks|key|1322425|text|在Java8中，使用CompletableFuture可以更好地实现这一点。假设我们有一个从数据库获取id的类，为了简单起见，我们可以像下面这样返回一个数字，|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|1322426|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;
++++}
}|code-block|syntax|javascript|1322427|现在，一旦每个将来的结果都准备好了，我们就可以将结果添加到并发收集中。|1322428|Collection<Integer>+results+=+new+ConcurrentLinkedQueue<>();
int+tasks+=+10;
CompletableFuture<?>[]+allFutures+=+new+CompletableFuture[tasks];
for+(int+i+=+0;+i+<+tasks;+i%2B%2B)+{
+++++int+temp+=+i;
+++++CompletableFuture<Integer>+future+=+CompletableFuture.supplyAsync(()->+new+GenerateNumber(temp).get(),+executor);
+++++allFutures[i]+=+future.thenAccept(results::add);
+}|1322429|现在我们可以在所有的期货都准备好的时候添加一个回调，|1322430|CompletableFuture.allOf(allFutures).thenAccept(c->{
+++System.out.println(results);+//+do+something+with+result
});|1322431|entityMap|0|LINK|mutability|MUTABLE|url|https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html^0|A|H|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|Y|8|@]|9|@$A|Z|B|10|1|11]]|C|$]]|$1|D|3|E|5|F|7|12|8|@]|9|@]|C|$G|H]]|$1|I|3|J|5|6|7|13|8|@]|9|@]|C|$]]|$1|K|3|L|5|F|7|14|8|@]|9|@]|C|$G|H]]|$1|M|3|N|5|6|7|15|8|@]|9|@]|C|$]]|$1|O|3|P|5|F|7|16|8|@]|9|@]|C|$G|H]]|$1|Q|3|-4|5|6|7|17|8|@]|9|@]|C|$]]]|R|$S|$5|T|U|V|C|$W|X]]]]

In Java8 there is better way for doing this using <a href="https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html" rel="nofollow">CompletableFuture</a>. Say we have class that get's id from the database, for simplicity we can just return a number as below,

<pre><code>static class GenerateNumber implements Supplier&lt;Integer&gt;{

 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;
 }
}
</code></pre>

Now we can add the result to a concurrent collection once the results of every future is ready.

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

Now we can add a callback when all the futures are ready,

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

I have a Java Thread like the following:

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

I have about 300 ids, and every couple of seconds - I fire up threads to make a call for each of the id. Eg.

<pre><code>for(String id: ids) {
 MyThread thread = new MyThread(id);
 thread.start();
}
</code></pre>

Now, I would like to collect the results from each threads, and do a batch insert to the database, instead of making 300 database inserts every 2 seconds.

Any idea how I can accomplish this?

Return values from Java Threads

数据库

Java

我有一个Java线程，如下所示：   public class MyThread extends Thread {        MyService service;        String id;        public MyThread(String id) {            this.id = n...

问从Java线程返回值
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问从Java线程返回值EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问从Java线程返回值
EN