solr导入数据的高效方法

   Solr提供了丰富的数据导入接口，可以导入数据库表、xml、json、csv各种格式的数据信息。

Solr的数据导入接口可以分为两类：DIH接口和HTTP接口 。关于DIH接口的用法可以参看链接：http://blog.chenlb.com/2010/03/solr-data-import-quick-start.html，我就不重复造轮子了。

关于HTTP接口有EmbeddedSolrServer、ConcurrentUpdateSolrServer、HttpSolrServer

其中EmbeddedSolrServer是不用走HTTP通道的，所以性能比其它两个SolrServer要高1.5倍。但是本质上讲，这三个SolrServer在导入数据的处理方式上是一致的。其处理的过程如下：

1、把数据信息封装成Collection<SolrInputDocument>。

2、把Collection<SolrInputDocument>转化成xml或者序列化成javabin。

3、形成SolrRequest，把转化后的信息封装成SolrRequest的参数。

4、如果是EmbeddedSolrServer，则直接对应到UpdateHandler。如果是HttpSolrServer,则走HTTP通道，需要用HttpClient把参数传递到服务器。

5、把xml或者javabin还原成SolrInputDocument。

6、把SolrInputDocument转化成lucene的Document。

7、用lucene的IndexWriter把Document写入到索引中。

这里面最让人费解的是在solrj中已经形成了SolrInputDocument，但是还得把SolrInputDocument转化成xml或者javabin，然后再还原回来。而且需要用Collection来保存。

而且，如果用EmbeddedSolrServer，无需走HTTP通道，也只能这样做。让人感到恼火。其实是有其它办法的。如果我们拿到了SolrCore，通过SolrCore就可以拿到UpdateProcessorChain，通过ProcessorChain就可以绕过前面的5步，直接到第6步了。

处理的代码如下：

public class SolrRecordHandler implements Runnable{
//生产者-消费者 solr doc
    private ArrayBlockingQueue<SolrInputDocument> docs=new ArrayBlockingQueue<SolrInputDocument>(5000);
        public void wrap(ResultSet rs){
SolrInputDocument doc=new SolrInputDocument();
        try {
             ResultSetMetaData rsm = rs.getMetaData();
             int numColumns = rsm.getColumnCount();
             for (int i = 1; i < (numColumns + 1); i++)
             {
                    doc.addField(rsm.getColumnName(i), rs.getObject(i));
             }
                                                                                                                                
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
        docs.add(doc);
        }
@Override
    public void run() {
        logger.info("solr 写线程启动开始。。。。");
        SolrCore core = cores.getCore("review");
        UpdateRequestProcessorChain chain=  core.getUpdateProcessingChain(null);
        SolrParams param=new ModifiableSolrParams();
        SolrQueryRequestBase req=new SolrQueryRequestBase(core,param){};
        SolrQueryResponse rsp=new SolrQueryResponse();
        UpdateRequestProcessor processor=chain.createProcessor(req, rsp);
        //不停地从队列中读取元素，直到任务结束
        SolrInputDocument doc;
        AddUpdateCommand acmd=new AddUpdateCommand(req);
        while(true){
            try {
                doc=docs.take();
                //读取到一个空的doc，则表明任务结束
                if(doc.isEmpty()){
                    break;
                }
                acmd.solrDoc=doc;
                processor.processAdd(acmd);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        logger.info("solr index thread finished!");
        //任务完成，则提交
        try {
            CommitUpdateCommand cmd=new CommitUpdateCommand(req, false);
            processor.processCommit(cmd);
        } catch (IOException e) {
            e.printStackTrace();
        }finally{
            try {
                processor.finish();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        isfinished.set(true);
    }
}

上面的代码是多线程的，一个线程负责把数据库的封装成SolrInputDocument，然后放到阻塞队列中，另外一个线程负责从阻塞队列中取出SolrInputDocument，然后添加到索引中。

最后，感谢@李雨前 的帮助。

本文出自 “每天进步一点点” 博客，请务必保留此出处http://sbp810050504.blog.51cto.com/2799422/1403656