java多线程处理执行solr创建索引示例

代码如下:

public class SolrIndexer implements Indexer, Searcher, DisposableBean {
 //~ Static fields/initializers =============================================

static final Logger logger = LoggerFactory.getLogger(SolrIndexer.class);

private static final long SHUTDOWN_TIMEOUT    = 5 * 60 * 1000L; // long enough

private static final int  INPUT_QUEUE_LENGTH  = 16384;

//~ Instance fields ========================================================

private CommonsHttpSolrServer server;

private BlockingQueue<Operation> inputQueue;

private Thread updateThread;
 volatile boolean running = true;
 volatile boolean shuttingDown = false;

//~ Constructors ===========================================================

public SolrIndexer(String url) throws MalformedURLException {
  server = new CommonsHttpSolrServer(url);

inputQueue = new ArrayBlockingQueue<Operation>(INPUT_QUEUE_LENGTH);

updateThread = new Thread(new UpdateTask());
  updateThread.setName("SolrIndexer");
  updateThread.start();
 }

//~ Methods ================================================================

public void setSoTimeout(int timeout) {
  server.setSoTimeout(timeout);
 }

public void setConnectionTimeout(int timeout) {
  server.setConnectionTimeout(timeout);
 }

public void setAllowCompression(boolean allowCompression) {
  server.setAllowCompression(allowCompression);
 }

public void addIndex(Indexable indexable) throws IndexingException {
  if (shuttingDown) {
   throw new IllegalStateException("SolrIndexer is shutting down");
  }
  inputQueue.offer(new Operation(indexable, OperationType.UPDATE));
 }

public void delIndex(Indexable indexable) throws IndexingException {
  if (shuttingDown) {
   throw new IllegalStateException("SolrIndexer is shutting down");
  }
  inputQueue.offer(new Operation(indexable, OperationType.DELETE));
 }

private void updateIndices(String type, List<Indexable> indices) throws IndexingException {
  if (indices == null || indices.size() == 0) {
   return;
  }

logger.debug("Updating {} indices", indices.size());

UpdateRequest req = new UpdateRequest("/" + type + "/update");
  req.setAction(UpdateRequest.ACTION.COMMIT, false, false);

for (Indexable idx : indices) {
   Doc doc = idx.getDoc();

SolrInputDocument solrDoc = new SolrInputDocument();
   solrDoc.setDocumentBoost(doc.getDocumentBoost());
   for (Iterator<Field> i = doc.iterator(); i.hasNext();) {
    Field field = i.next();
    solrDoc.addField(field.getName(), field.getValue(), field.getBoost());
   }

req.add(solrDoc);   
  }

try {
   req.process(server);   
  } catch (SolrServerException e) {
   logger.error("SolrServerException occurred", e);
   throw new IndexingException(e);
  } catch (IOException e) {
   logger.error("IOException occurred", e);
   throw new IndexingException(e);
  }
 }

private void delIndices(String type, List<Indexable> indices) throws IndexingException {
  if (indices == null || indices.size() == 0) {
   return;
  }

logger.debug("Deleting {} indices", indices.size());

UpdateRequest req = new UpdateRequest("/" + type + "/update");
  req.setAction(UpdateRequest.ACTION.COMMIT, false, false);
  for (Indexable indexable : indices) {   
   req.deleteById(indexable.getDocId());
  }

try {
   req.process(server);
  } catch (SolrServerException e) {
   logger.error("SolrServerException occurred", e);
   throw new IndexingException(e);
  } catch (IOException e) {
   logger.error("IOException occurred", e);
   throw new IndexingException(e);
  }
 }

public QueryResult search(Query query) throws IndexingException {
  SolrQuery sq = new SolrQuery();
  sq.setQuery(query.getQuery());
  if (query.getFilter() != null) {
   sq.addFilterQuery(query.getFilter());
  }
  if (query.getOrderField() != null) {
   sq.addSortField(query.getOrderField(), query.getOrder() == Query.Order.DESC ? SolrQuery.ORDER.desc : SolrQuery.ORDER.asc);
  }
  sq.setStart(query.getOffset());
  sq.setRows(query.getLimit());

QueryRequest req = new QueryRequest(sq);
  req.setPath("/" + query.getType() + "/select");

try {
   QueryResponse rsp = req.process(server);
   SolrDocumentList docs = rsp.getResults();

QueryResult result = new QueryResult();
   result.setOffset(docs.getStart());
   result.setTotal(docs.getNumFound());
   result.setSize(sq.getRows());

List<Doc> resultDocs = new ArrayList<Doc>(result.getSize());
   for (Iterator<SolrDocument> i = docs.iterator(); i.hasNext();) {
    SolrDocument solrDocument = i.next();

Doc doc = new Doc();
    for (Iterator<Map.Entry<String, Object>> iter = solrDocument.iterator(); iter.hasNext();) {
     Map.Entry<String, Object> field = iter.next();
     doc.addField(field.getKey(), field.getValue());
    }

resultDocs.add(doc);
   }

result.setDocs(resultDocs);
   return result;

} catch (SolrServerException e) {
   logger.error("SolrServerException occurred", e);
   throw new IndexingException(e);
  }
 }

public void destroy() throws Exception {
  shutdown(SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS);  
 }

public boolean shutdown(long timeout, TimeUnit unit) {
  if (shuttingDown) {
   logger.info("Suppressing duplicate attempt to shut down");
   return false;
  }
  shuttingDown = true;
  String baseName = updateThread.getName();
  updateThread.setName(baseName + " - SHUTTING DOWN");
  boolean rv = false;
  try {
   // Conditionally wait
   if (timeout > 0) {
    updateThread.setName(baseName + " - SHUTTING DOWN (waiting)");
    rv = waitForQueue(timeout, unit);
   }
  } finally {
   // But always begin the shutdown sequence
   running = false;
   updateThread.setName(baseName + " - SHUTTING DOWN (informed client)");
  }
  return rv;
 }

/**
  * @param timeout
  * @param unit
  * @return
  */
 private boolean waitForQueue(long timeout, TimeUnit unit) {
  CountDownLatch latch = new CountDownLatch(1);
  inputQueue.add(new StopOperation(latch));
  try {
   return latch.await(timeout, unit);
  } catch (InterruptedException e) {
   throw new RuntimeException("Interrupted waiting for queues", e);
  }
 }

class UpdateTask implements Runnable {
  public void run() {
   while (running) {
    try {
     syncIndices();
    } catch (Throwable e) {
     if (shuttingDown) {
      logger.warn("Exception occurred during shutdown", e);
     } else {
      logger.error("Problem handling solr indexing updating", e);
     }
    }
   }
   logger.info("Shut down SolrIndexer");
  }
 }

void syncIndices() throws InterruptedException {
  Operation op = inputQueue.poll(1000L, TimeUnit.MILLISECONDS);

if (op == null) {
   return;
  }

if (op instanceof StopOperation) {
   ((StopOperation) op).stop();
   return;
  }

// wait 1 second
  try {
   Thread.sleep(1000);
  } catch (InterruptedException e) {

}

List<Operation> ops = new ArrayList<Operation>(inputQueue.size() + 1);
  ops.add(op);
  inputQueue.drainTo(ops);

Map<String, List<Indexable>> deleteMap = new HashMap<String, List<Indexable>>(4);
  Map<String, List<Indexable>> updateMap = new HashMap<String, List<Indexable>>(4);

for (Operation o : ops) {
   if (o instanceof StopOperation) {
    ((StopOperation) o).stop();
   } else {
    Indexable indexable = o.indexable;
    if (o.type == OperationType.DELETE) {
     List<Indexable> docs = deleteMap.get(indexable.getType());
     if (docs == null) {
      docs = new LinkedList<Indexable>();
      deleteMap.put(indexable.getType(), docs);
     }
     docs.add(indexable);
    } else {
     List<Indexable> docs = updateMap.get(indexable.getType());
     if (docs == null) {
      docs = new LinkedList<Indexable>();
      updateMap.put(indexable.getType(), docs);
     }
     docs.add(indexable);
    }
   }
  }

for (Iterator<Map.Entry<String, List<Indexable>>> i = deleteMap.entrySet().iterator(); i.hasNext();) {
   Map.Entry<String, List<Indexable>> entry = i.next();
   delIndices(entry.getKey(), entry.getValue());
  }

for (Iterator<Map.Entry<String, List<Indexable>>> i = updateMap.entrySet().iterator(); i.hasNext();) {
   Map.Entry<String, List<Indexable>> entry = i.next();
   updateIndices(entry.getKey(), entry.getValue());
  }
 }

enum OperationType { DELETE, UPDATE, SHUTDOWN }

static class Operation {
  OperationType type;
  Indexable indexable;

Operation() {}

Operation(Indexable indexable, OperationType type) {
   this.indexable = indexable;
   this.type = type;
  }
 }

static class StopOperation extends Operation {
  CountDownLatch latch;

StopOperation(CountDownLatch latch) {
   this.latch = latch;
   this.type = OperationType.SHUTDOWN;
  }

public void stop() {
   latch.countDown();
  }
 }

//~ Accessors ===============

}