我想使用ExecutorService来运行一系列相同的Runnable / Callable任务 . 我一直在寻找一个教程或一个例子,但没有涉及实际设置现有Runnable / Callable对象的值,然后使用submit()将该对象发送回ExecutorService .
基本上,这就是我想要做的:
-
获取服务器列表 .
-
通过服务器列表迭代,调用
InetAddress.getByName(host)
以获取每个主机上的数据 . -
将数据收集到服务器bean中以存储在数据库中 .
所以,现在,使用10,000()服务器,它需要永远 . 所以,我的想法是使用ExecutorService来管理线程池 . 我似乎无法弄清楚的是如何检测一个线程何时完成,以便我可以获取数据 . 然后我需要获取列表中的下一个服务器,将其放入Task中,然后将()提交回ExecutorService .
也就是说,到目前为止我所读到的内容似乎指向以下内容:ExecutorService,submit(),Callable,Future .
所以,作为伪造的过程:
-
获取服务器列表 .
-
使用numThreads线程数设置ExecutorService
-
迭代numThreads并创建numThreads WorkerTask()对象 .
-
Submit()WorkerTask()到ExecutorService进行处理 .
-
检测WorkerTask()何时完成,获取Callable(Future)结果 .
-
获取下一台服务器 .
-
将服务器值设置为WorkerTask()< - 如何?这是难以捉摸的......
-
将WorkerTask()(带有新值)提交给ExecutorService .
-
再次迭代 .
-
ExecutorService.shutdown()...
因此,这个教程或示例会很棒...尤其是将新值放入WorkerTask()的示例 . 此外,我对这个提议的解决方案有任何想法吗?这是坏的,好的,或者如果有另一种方式,我是开放的 .
02/09/2014 - 编辑和添加
嗨,所以以下是第一次切入 . 但回答提出的一些问题:
- 我已经解决了将新数据放入Worker并重新提交到ExecutorService的问题...请参阅代码片段 .
- 我也解决了"get the stuff"的问题......我只是将Future()结果转换为Worker类...请参阅代码片段 .
- 最后,虽然我可以将每个服务器分配给Worker()和Future(),但我担心当前10,000会增长并且内存将成为问题 .
也就是说,这是第一次尝试,这非常有效 . 运行速度更快,只使用getNumbnerThreads()Worker和Future对象:
public List<ServerBean> lookupHostIps ( List<ServerBean> theServerList ) {
//ServerBean serverDto = null;
ServerBean ipDto = null;
List<ServerBean> theResults = new ArrayList<ServerBean>();
List<HostLookupWorker> theWorkers = new ArrayList<HostLookupWorker>( getNumberThreads() );
List<Future<HostLookupWorker>> theFutures = new ArrayList<Future<HostLookupWorker>>( getNumberThreads() );
ExecutorService executor = Executors.newFixedThreadPool ( getNumberThreads() );
// WORKERS : Create the workers...prime them with a server
// bean...
//
for (int j = 0; j < getNumberThreads(); j++) {
//for (int j = 0; j < theServerList.size(); j++) {
theWorkers.add ( new HostLookupWorker( theServerList.get(j) ) );
Future<HostLookupWorker> theFuture = executor.submit ( theWorkers.get ( j ) );
theFutures.add ( j, theFuture );
}
int lloopItems = getNumberThreads(); /* loops thru all servers */
//int lloopThreads = 0; /* loops thru threads */
int lidxThread = 0; /* what thread is ready */
//int lidxFuture = 0; /* what future is ready */
boolean lblnNext = false; /* is a thread done/ready */
int lidxWorkers = 0; /* tracks the futures */
while ( lloopItems < theServerList.size() ) {
// READY : Is one of the threads ready for more work?
if ( lblnNext ) {
// VALUE : Grab the thread by index and set the next
// server value.
theWorkers.get ( lidxThread ).setBean ( theServerList.get(lloopItems) );
getLog().debug ( "Thread [" + lidxThread + "] Assigned Host ["+theServerList.get(lloopItems).getServerName ()+"] " );
// FUTURE : Package a new Future<HostLookupWorker>
// and submit it to the thread pool.
Future<HostLookupWorker> theFuture = executor.submit ( theWorkers.get ( lidxThread ) );
theFutures.add ( lidxThread, theFuture );
lblnNext = false; /* reset to allow for another thread */
lloopItems++; /* increment the main loop counter */
}
while ( !(lblnNext) ) {
try {
if ( theFutures.get(lidxWorkers).get() != null ) {
// GET THE STUFF : Grab the results from the Future...
HostLookupWorker ltheItem = theFutures.get(lidxWorkers).get();
if ( ltheItem.getValue () != null ) {
if (!ltheItem.getValue ().contains("Cannot find host")){
ipDto = new ServerBean ();
ipDto.setServerId ( ltheItem.getBean ().getServerId() );
ipDto.setServerName ( ltheItem.getBean ().getServerName() );
ipDto.setIpAddress ( ltheItem.getValue () );
theResults.add(ipDto);
}
lidxThread = lidxWorkers; /* this thread is ready for more work */
lblnNext = true; /* flag the upper condition to assign new work */
getLog().debug ( "Thread [" + lidxThread + "] Host ["+ltheItem.getHost ()+"] has IP ["+ltheItem.getValue()+"]" );
}
}
else {
getLog().debug ( "Thread [" + lidxThread + "] NULL" );
}
lidxWorkers++; /* next worker/future */
if ( lidxWorkers >= getNumberThreads() ) {
lidxWorkers = 0;
}
}
catch(ExecutionException e){
getLog().error ( e );
}
catch(InterruptedException e){
getLog().error ( e );
}
}
}
executor.shutdown ();
return theResults;
}
这是Worker / Thread类:
import java.net.*;
import java.util.concurrent.Callable;
import com.lmig.cdbatch.dto.ServerBean;
public class HostLookupWorker implements Callable {
private InetAddress node = null;
private String value = null;
private boolean busy = false;
private ServerBean bean = null;
public HostLookupWorker () {
this.busy = false;
}
// public HostLookupWorker ( String theHost ) {
// this.busy = false;
// this.host = theHost;
// }
public HostLookupWorker ( ServerBean theItem ) {
this.busy = false;
this.bean = theItem;
//this.host = theItem.getServerName ().trim ();
}
public String lookup ( String host ) {
if ( host != null ) {
// get the bytes of the IP address
try {
this.node = InetAddress.getByName ( host );
}
catch ( UnknownHostException ex ) {
this.value = "Not Found [" + getHost() + "]";
return "Not Found [" + host + "]";
}
if ( isHostname(host) ) {
getBean().setIpAddress ( node.getHostAddress() );
return node.getHostAddress();
}
else { // this is an IP address
//return node.getHostName();
return host;
}
}
return host;
} // end lookup
public boolean isHostname(String host) {
// Is this an IPv6 address?
if (host.indexOf(':') != -1)
return false;
char[] ca = host.toCharArray();
// if we see a character that is neither a digit nor a period
// then host is probably a hostname
for (int i = 0; i < ca.length; i++) {
if (!Character.isDigit(ca[i])) {
if (ca[i] != '.')
return true;
}
}
// Everything was either a digit or a period
// so host looks like an IPv4 address in dotted quad format
return false;
} // end isHostName
// public void run() {
// value = lookup ( getHost() );
//
// }
public Object call() throws Exception {
Thread.sleep ( 10000 );
this.busy = true;
this.value = lookup ( getHost() );
this.busy = false;
return this;
}
public String getHost() {
return getBean().getServerName ().trim ();
}
public void setHost(String host) {
getBean().setServerName ( host.trim () );
}
public InetAddress getNode() {
return node;
}
public void setNode(InetAddress node) {
this.node = node;
}
public String getValue() {
return value;
}
public void setValue(String value) {
this.value = value;
}
public boolean isBusy() {
return busy;
}
public void setBusy(boolean busy) {
this.busy = busy;
}
public ServerBean getBean() {
return bean;
}
public void setBean(ServerBean bean) {
this.bean = bean;
}
}
所以,总结一下:
- 该过程确实有效,并且运行速度很快 .
- 我需要修改一下代码,因为有getNumberThreads() - 当更大的while()循环结束时,1期货未经处理...
那么,我现在正在努力的是如何检测线程何时完成...我已经看到了多个示例,一个测试Future()!= null,其他测试Future()== null . 那么哪一个是对的?
1 回答
我认为在这种情况下最好的方法是为每个服务器创建一个任务,因为它们将由pull中的线程执行,然后使用tge future对象来检索任务返回的服务器信息 .