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

复制代码 代码如下:

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 ===============

}

时间: 2024-09-22 10:40:33

java多线程处理执行solr创建索引示例_java的相关文章

关于solr创建索引的效率

问题描述 关于solr创建索引的效率 单核solr服务器,有分词,一个document有40+个field,使用solrj创建索引的速度能够达到多少条每秒? 或者影响solr建索引的效率的因素有哪些? 解决方案 http://www.cnblogs.com/likehua/archive/2015/04/29/4465514.html

java多线程和并发包入门示例_java

一.java多线程基本入门java多线程编程还是比较重要的,在实际业务开发中经常要遇到这个问题. java多线程,传统创建线程的方式有两种. 1.继承自Thread类,覆写run方法. 2.实现Runnable接口,实现run方法. 启动线程的方法都是调用start方法,真正执行调用的是run方法.参考代码如下: 复制代码 代码如下: package com.jack.thread; /** * 线程简单演示例子程序 *  * @author pinefantasy * @since 2013-

java压缩文件和下载图片示例_java

本文实例为大家分享了java压缩文件和下载图片示例,供大家参考,具体内容如下 主页面index.xml <%@ page language="java" import="java.util.*" pageEncoding="utf-8"%> <html> <head> <title>项目的主页</title> </head> <body> <h2>主页

java跟踪执行的sql语句示例分享_java

代码: 复制代码 代码如下: package com.lwj.test.proxy; import java.lang.reflect.InvocationHandler;import java.lang.reflect.InvocationTargetException;import java.lang.reflect.Method;import java.lang.reflect.Proxy;import java.sql.Connection;import java.sql.SQLExce

Java实现Map集合二级联动示例_java

Map集合可以保存键值映射关系,这非常适合本实例所需要的数据结构,所有省份信息可以保存为Map集合的键,而每个键可以保存对应的城市信息,本实例就是利用Map集合实现了省市级联选择框,当选择省份信息时,将改变城市下拉选择框对应的内容. 思路分析: 1. 创建全国(省,直辖市,自治区)映射集合,即LinkedHashMap对象,使用Map接口的put()方法向集合中添加指定的省与城市的映射关系,其中值为String型一维数组. 代码如下: CityMap.java 复制代码 代码如下: import

JAVA实现空间索引编码——GeoHash的示例_java

之前自己在做基于Lucene的内容检索过程中,了解到Lucene可以实现对文本信息,数值信息的内容检索,对于空间距离好像并为为源码中实现:最近半年自己接触到Solr,里面有一个空间距离检索(经纬度),最近对其中的实现做了下学习,了解到在实现空间距离检索的有一个比较常用的技术--GeoHash,下面就介绍下GeoHash. GeoHash特点 GeoHash用一个字符串表示经度和纬度两个坐标,比如我现在所在位置的GeoHash值为 wx4sv61q: GeoHash标识的并不是一个点,而是一个区域

浅析Java中线程的创建和启动_java

前言 线程是一个程序内部的顺序控制流.cpu实际上在一个时间点上,只执行一个.只不过我们把cup分成了多个时间片,由于速度很快,我们看起来像是多个线程.就像你的时间分成几片,这样整体看来做事情有规律,效率就会高,何况是cup呢. 线程的创建和启动:      (1)定义Thread类的子类,并重写了该类的run()方法,该run()方法的方法就代表了线程需要完成的任务.因此把run()方法称为线程执行体      (2)创建Thread子类的实例,即创建了线程对象      (3)调用线程对象的

Java模拟有序链表数据结构的示例_java

有序链表:按关键值排序.删除链头时,就删除最小(/最大)的值,插入时,搜索插入的位置. 插入时需要比较O(N),平均O(N/2),删除最小(/最大)的在链头的数据时效率为O(1), 如果一个应用需要频繁的存取(插入/查找/删除)最小(/最大)的数据项,那么有序链表是一个不错的选择 优先级队列 可以使用有序链表来实现 有序链表的插入排序: 对一个无序数组,用有序链表来排序,比较的时间级还是O(N^2) 复制时间级为O(2*N),因为复制的次数较少,第一次放进链表数据移动N次,再从链表复制到数组,又

java使用httpclient发送post请求示例_java

复制代码 代码如下: package org.ssi.util; import java.io.IOException;import java.util.ArrayList;import java.util.List; import net.sf.json.JSONArray; import org.apache.commons.lang.exception.ExceptionUtils;import org.apache.commons.logging.Log;import org.apach