标签:des style c class blog code
实现索引更新的异步处理,以及查询接口,日志/线程池/队列监控没有加上。
接口:
public interface ISolrDocument { public SolrInputDocument convertToInputDocument() throws Exception; public void buildSolrDocument(SolrDocument document) throws Exception; }
实现:
public class SolrDocumentImpl implements ISolrDocument { @Override public SolrInputDocument convertToInputDocument() throws Exception{ final SolrInputDocument doc = new SolrInputDocument(); Field[] fields = this.getClass().getDeclaredFields(); for (Field f : fields) { f.setAccessible(true); String idxName = f.getName(); Object fieldValue =f.get(this); doc.addField(idxName, fieldValue); } return doc; } @Override public void buildSolrDocument(SolrDocument document) throws Exception{ Field[] fields = this.getClass().getDeclaredFields(); for (Field f : fields) { f.setAccessible(true); String idxName = f.getName(); Object fieldValue = document.get(idxName); f.set(this, fieldValue); } }
public class DataResult<T> { private int start; private int limits; private long totalRecord; private List<T> records; public int getLimits() { return limits; } public void setLimits(int limits) { this.limits = limits; } public long getTotalRecord() { return totalRecord; } public void setTotalRecord(long totalRecord) { this.totalRecord = totalRecord; } public List<T> getRecords() { return records; } public void setRecords(List<T> records) { this.records = records; } public int getStart() { return start; } public void setStart(int start) { this.start = start; } }
接口:
public interface ISolrServer { public ThreadPoolExecutor getThreadPool(); void init() throws Exception; /** * 索引优化 */ void optimize(); /** * 清理索引 * @param query */ void clearDocs(SolrQuery query); /** * 批量更新索引 * * @param docs */ void batchUpdateDocs(List<ISolrDocument> docs); /** * 分页查询 * * @param clazz * @param query * @return */ public <T extends ISolrDocument> DataResult<T> pageSearch(SolrQuery query, Class<T> clazz); void destory() throws Exception; }
实现:
public class SolrServerImpl implements ISolrServer { private CloudSolrServer solrServer; private String zookeeperStr; private String coreName; private String idName; private int corePoolSize; private int maxPoolSize; private int threadQueueSize; private ThreadPoolExecutor threadPool; @Override public void init() throws Exception { this.threadPool = new ThreadPoolExecutor(corePoolSize, maxPoolSize, 10000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(threadQueueSize), new SolrThreadFactory("solr-pool"), new ThreadPoolExecutor.DiscardOldestPolicy()); solrServer=new CloudSolrServer(this.zookeeperStr); solrServer.setDefaultCollection(coreName); solrServer.setIdField(idName); solrServer.setZkClientTimeout(10000); solrServer.setZkConnectTimeout(5000); } @Override public void optimize() { } @Override public void clearDocs(final SolrQuery query) { this.threadPool.execute(new Runnable() { @Override public void run() { try { SolrServerImpl.this.solrServer.deleteByQuery(query.getQuery()); SolrServerImpl.this.solrServer.commit(false, false, true); }catch (Exception e){ e.printStackTrace(); } } }); } @Override public void batchUpdateDocs(List<ISolrDocument> docs) { try { List<SolrInputDocument> inputDocs=new ArrayList<SolrInputDocument>(); for(ISolrDocument doc:docs){ inputDocs.add(doc.convertToInputDocument()); } this.solrServer.add(inputDocs); this.solrServer.commit(false,false,true); }catch (Exception e){ e.printStackTrace(); } } @Override public <T extends ISolrDocument> DataResult<T> pageSearch(SolrQuery query, Class<T> clazz) { int limit=query.getRows()==0?20:query.getRows(); int start=query.getStart(); List<T> dataResults=new ArrayList<T>(limit); DataResult datas=new DataResult(); datas.setLimits(limit); datas.setStart(start); try{ QueryResponse response= this.solrServer.query(query); SolrDocumentList solrDocumentList=response.getResults(); datas.setTotalRecord(solrDocumentList.getNumFound()); for (SolrDocument doc :solrDocumentList){ T iSolrDocument=clazz.newInstance(); iSolrDocument.buildSolrDocument(doc); dataResults.add(iSolrDocument); } datas.setRecords(dataResults); }catch (Exception e){ e.printStackTrace(); } return datas; } @Override public void destory() throws Exception { this.solrServer.shutdown(); } public void setIdName(String idName) { this.idName = idName; } public void setCoreName(String coreName) { this.coreName = coreName; } public void setZookeeperStr(String zookeeperStr) { this.zookeeperStr = zookeeperStr; } public CloudSolrServer getSolrServer() { return solrServer; } public void setSolrServer(CloudSolrServer solrServer) { this.solrServer = solrServer; } public String getZookeeperStr() { return zookeeperStr; } public String getCoreName() { return coreName; } public String getIdName() { return idName; } public int getCorePoolSize() { return corePoolSize; } public void setCorePoolSize(int corePoolSize) { this.corePoolSize = corePoolSize; } public int getMaxPoolSize() { return maxPoolSize; } public void setMaxPoolSize(int maxPoolSize) { this.maxPoolSize = maxPoolSize; } public int getThreadQueueSize() { return threadQueueSize; } public void setThreadQueueSize(int threadQueueSize) { this.threadQueueSize = threadQueueSize; } public ThreadPoolExecutor getThreadPool() { return threadPool; } public void setThreadPool(ThreadPoolExecutor threadPool) { this.threadPool = threadPool; } }
public class BatchTaskExecutor{ private ISolrServer iSolrServer; private int batchSize = 100; private int queueLength = 2000; private int workThreadSize = 2; private Semaphore semaphore; private LinkedBlockingQueue<ISolrDocument> docQueue; public void init(){ this.semaphore = new Semaphore(this.workThreadSize); docQueue = new LinkedBlockingQueue<ISolrDocument>(this.queueLength); while(true){ this.iSolrServer.getThreadPool().execute(new Runnable() { @Override public void run() { List<ISolrDocument> list=BatchTaskExecutor.this.batchPollDocs(); BatchTaskExecutor.this.acquire(); try { BatchTaskExecutor.this.iSolrServer.batchUpdateDocs(list); }catch (Exception e){ e.printStackTrace(); }finally { BatchTaskExecutor.this.release(); } } }); } } public boolean acquire() { try { return this.semaphore.tryAcquire(1,10, TimeUnit.SECONDS); } catch (InterruptedException e) { e.printStackTrace(); return false; } } public void release() { this.semaphore.release(); } public boolean offerDoc(ISolrDocument doc) { try { return this.docQueue.offer(doc, 1000, TimeUnit.MILLISECONDS); } catch (Exception e) { e.printStackTrace(); } return false; } public List<ISolrDocument> batchPollDocs() { List<ISolrDocument> docs = new ArrayList<ISolrDocument>(this.batchSize); while (docs.size() < this.batchSize) { ISolrDocument doc = null; try { doc = this.docQueue.poll(1000, TimeUnit.MILLISECONDS); } catch (Exception e) { e.printStackTrace(); } finally { if (doc==null) { break; } else { docs.add(doc); } } } return docs; } }
封装一个简单的solrserver组件,布布扣,bubuko.com
标签:des style c class blog code
原文地址:http://www.cnblogs.com/requelqi/p/3754871.html