图文精华

About云»专题 › 技术学习(版主发帖区) › 大数据学习 › Solr|Nutch|Lucene|ES › 实操演练› SolrJ 操作HttpSolrServer, ConcurrentUpdateSolrServer ...
分享

SolrJ 操作HttpSolrServer, ConcurrentUpdateSolrServer和CloudSolrServer

gefieder 2015-5-5 21:52:14 发表于 实操演练 [显示全部楼层] 只看大图 回帖奖励 阅读模式 关闭右栏 0 16249
问题导读：
1、如何实现SolrJ线程池连接数据库？
2、如何实现SolrJ的CRUD操作？




HttpSolrServer 使用了Apache Commons HTTP客户端来连接Solr. 注意在Solr 4.x中， CommonsHttpSolrServer已经改变为HttpSolrServer以及StreamingUpdateSolrServer已经改变为ConcurrentUpdateSolrServer 。ConcurrentUpdateSolrServer更适合update 操作，而HttpSolrServer 更适合query操作。
       添加document或是修改document。假如这个document已经存在，就会update这个document。代码片段如下:
  1. public void indexDocs() throws IOException, SolrServerException {  

  2.       server.setParser(new XMLResponseParser());  

  3.   

  4.       //Adds the docs and commit them.  

  5.       Collection<SolrInputDocument> docs = new LinkedList<SolrInputDocument>();  

  6.       /* i is used as identification of a document, which is treated as unique key.*/  

  7.       SolrInputDocument doc2 ;  

  8.       /*一千条数据，花费大约一小时，生产660M。使用多线程并发执行估计更好的*/  

  9.       for(int i =10000000; i < 10000002; i++){  

  10.           doc2 = new SolrInputDocument();  

  11.           doc2.addField("customer_id", i);  

  12.           doc2.addField("name", "John Natch-" + i);  

  13.           doc2.addField("level", "VIP");  

  14.           doc2.addField("sex", "男");  

  15.           doc2.addField("address", "【【【【【金刚金刚金刚金刚金刚金】】】】" + i);  

  16.           System.out.println("add doc "+ i);  

  17.           docs.add(doc2);  

  18.           if(docs.size() == 1000){  

  19.               server.add(docs);  

  20.               server.commit();  

  21.               logger.info("commit 1000 doc "+ i);  

  22.               docs.clear();  

  23.           }  

  24.           /* 

  25.           To immediately commit after adding documents, you could use: 

  26.  

  27.            UpdateRequest req = new UpdateRequest(); 

  28.            req.setAction( UpdateRequest.ACTION.COMMIT, false, false ); 

  29.            req.add( docs ); 

  30.            UpdateResponse rsp = req.process( server ); 

  31.            */  

  32.       }  

  33.       server.add(docs);  

  34.       server.commit();  

  35.       logger.info("Commits successfully!......");  

  36.   }
复制代码
        能够执行代码前，在Solr core的配置文件shema.xml中配置具体的字段。
  1.  <!-- core  'customer' schema field definition -->  

  2.   

  3. <field name="customer_id" type="int" indexed="true" stored="true" required="true"  multiValued="false"/>   

  4. <field name="name" type="string" indexed="true" stored="true"/>  

  5. <field name="sex" type="string" indexed="true" stored="false"/>  

  6. <field name="level" type="string" indexed="true" stored="true"/>  

  7. <field name="address" type="string" indexed="true" multiValued="true" stored="true"/>
复制代码
  1. <uniqueKey>customer_id</uniqueKey>  
复制代码


        删除操作：
  1. private void commitDocs(Collection<SolrInputDocument> docs){  

  2.     try {  

  3.         //server.deleteById(1)    //specify the id list you want to be deleted.  

  4.         server.add(docs);  

  5.         server.commit();  

  6.         docs.clear();  

  7.     } catch (SolrServerException e) {  

  8.         logger.error("SolrServerException", e);  

  9.     } catch (IOException e) {  

  10.         logger.error("IOException", e) ;  

  11.     }  

  12. }
复制代码
      与数据集成，实现使用SolrJ操作数据库。当然，这个可以使用Solr DIH实现。两种各有其优缺点，根据实际的应用来选择具体的实现方式。
  1. public void indexDocsWithDB(){  

  2.        PoolingDataSourceDemo dataSource = new PoolingDataSourceDemo();  

  3.        List<List<Object>>  rows = dataSource.executeQuerySQL("select * from customer");  

  4.        String[]  columnNames = dataSource.getColNames();  

  5.        Collection<SolrInputDocument> docs = new LinkedList<SolrInputDocument>();  

  6.        SolrInputDocument doc ;  

  7.        for(List row : rows)  {  

  8.            int size = row.size() + 1;  

  9.            doc = new SolrInputDocument();  

  10.            for(int i = 1; i < size ; i++){  

  11.                doc.addField(columnNames[i], row.get(i-1)) ;  

  12.                logger.info(columnNames[i]+"add filed "+ row.get(i-1)) ;  

  13.            }  

  14.            docs.add(doc);  

  15.            if(docs.size() > 100){  

  16.                commitDocs(docs);  

  17.            }  

  18.        }  

  19.        if(docs.size() > 0){  

  20.            commitDocs(docs);  

  21.        }  

  22.    }
复制代码
     完整的代码：
    PoolingDataSourceDemo.java   实现线程池连接数据库。
  1. import net.spy.memcached.compat.log.Logger;  

  2. import net.spy.memcached.compat.log.LoggerFactory;  

  3. import org.apache.commons.dbcp.*;  

  4. import org.apache.commons.pool.impl.GenericObjectPool;  

  5.   

  6. import javax.sql.DataSource;  

  7. import java.sql.*;  

  8. import java.util.ArrayList;  

  9. import java.util.LinkedList;  

  10. import java.util.List;  

  11.   

  12. /** 

  13.  * @author John Liu 

  14.  * @see 

  15.  */  

  16. public class PoolingDataSourceDemo {  

  17.   

  18.     private final static Logger logger = LoggerFactory.getLogger(PoolingDataSourceDemo.class) ;  

  19.     /* These properties can be configured in a properties type file*/  

  20.     private final static String CONNECTION_URL =  "jdbc:mysql://localhost/pythondb?autoReconnect=true";  

  21.     private final static String DRIVER_CLASS = "com.mysql.jdbc.Driver";  

  22.     private final static String USER_NAME = "elite";  

  23.     private final static String PASSWORD = "elite";  

  24.   

  25.     private final static int MAX_ACTIVE_NUMBER = 10;  

  26.   

  27.     private static GenericObjectPool connectionPool = null;  

  28.   

  29.     private String[] colNames ;  

  30.   

  31.   

  32.   

  33.     private static DataSource dataSource;  

  34.   

  35.     static {  

  36.         dataSource  =  initDataSource();  

  37.     }  

  38.   

  39.     public GenericObjectPool getConnectionPool() {  

  40.         return connectionPool;  

  41.     }  

  42.   

  43.     public  List<List<Object>>  executeQuerySQL(String querySQL){  

  44.         Connection conn = null;  

  45.         Statement stmt = null;  

  46.         ResultSet resultSet = null;  

  47.         List<List<Object>> result = new LinkedList<List<Object>>();  

  48.         try {  

  49.             logger.info("Creating connection.");  

  50.             conn = dataSource.getConnection();  

  51.             stmt = conn.createStatement();  

  52.             resultSet = stmt.executeQuery(querySQL);  

  53.             //show the connection pool status  

  54.             printDataSourceStats();  

  55.             logger.info("Results:");  

  56.             int columnCount = resultSet.getMetaData().getColumnCount();  

  57.   

  58.             ResultSetMetaData rsm = resultSet.getMetaData();  

  59.             colNames = new String[columnCount + 1];  

  60.             for (int i = 1; i < (columnCount + 1); i++) {  

  61.                 colNames[i] = rsm.getColumnName(i).toLowerCase();  

  62.                 logger.info("column name: "+ colNames[i]) ;  

  63.             }  

  64.             List<Object> list ;  

  65.             while(resultSet.next()) {  

  66.                 list = new ArrayList<Object>() ;  

  67.                 for(int i=1; i<= columnCount; i++) {  

  68.                     Object obj =  getColumnValue(rsm, resultSet, colNames, i);  

  69.                     list.add(obj)  ;  

  70.                 }  

  71.                 result.add(list);  

  72.             }  

  73.         } catch(SQLException e) {  

  74.             e.printStackTrace();  

  75.             shutdownDataSource(dataSource);  

  76.         } finally {  

  77.             try { if (resultSet != null) resultSet.close(); } catch(Exception e) { }  

  78.             try { if (stmt != null) stmt.close(); } catch(Exception e) { }  

  79.             try { if (conn != null) conn.close(); } catch(Exception e) { }  

  80.             logger.info("result size: "+ result.size());  

  81.             return result;  

  82.         }  

  83.     }  

  84.   

  85.     public Object getColumnValue(ResultSetMetaData rsm, ResultSet rs, String[] colNames, int j) throws SQLException {  

  86.         Object f = null;  

  87.         if (colNames[j] != null) {  

  88.             switch (rsm.getColumnType(j)){  

  89.                 case Types.BIGINT:{  

  90.                     f = rs.getLong(j);  

  91.                     break;  

  92.                 }  

  93.                 case Types.INTEGER: {  

  94.                     f = rs.getInt(j);  

  95.                     break;  

  96.                 }  

  97.                 case Types.DATE:{  

  98.                     f = rs.getDate(j);  

  99.                     break;  

  100.                 }  

  101.                 case Types.FLOAT:{  

  102.                     f = rs.getFloat(j);  

  103.                     break;  

  104.                 }  

  105.                 case Types.DOUBLE:{  

  106.                     f = rs.getDouble(j);  

  107.                     break;  

  108.                 }  

  109.                 case Types.TIME: {  

  110.                     f = rs.getDate(j);  

  111.                     break;  

  112.                 }  

  113.                 case Types.BOOLEAN:{  

  114.                     f = rs.getBoolean(j);  

  115.                     break;  

  116.                 }  

  117.                 default:{  

  118.                     f = rs.getString(j);  

  119.                 }  

  120.             }  

  121.         }  

  122.         logger.info("column value: "+ f)  ;  

  123.         return f;  

  124.     }  

  125.     /** 

  126.      * [mysql] 

  127.      * #hibernate.connection.driver_class com.mysql.jdbc.Driver 

  128.      #hibernate.connection.url jdbc:mysql:///test 

  129.      #hibernate.connection.username gavin 

  130.      #hibernate.connection.password 

  131.      * @return DataSource 

  132.      */  

  133.     public static DataSource initDataSource(){  

  134.         //  

  135.         // Load JDBC Driver class.  

  136.         //  

  137.         try {  

  138.             Class.forName(DRIVER_CLASS).newInstance();  

  139.         } catch (InstantiationException e) {  

  140.             logger.error("InstantiationException error", e);  

  141.         } catch (IllegalAccessException e) {  

  142.             logger.error("IllegalAccessException error", e);  

  143.         } catch (ClassNotFoundException e) {  

  144.             logger.error("ClassNotFoundException error", e);  

  145.         }  

  146.   

  147.         //  

  148.         // Creates an instance of GenericObjectPool that holds our  

  149.         // pool of connections object.  

  150.         //  

  151.         connectionPool = new GenericObjectPool();  

  152.         connectionPool.setMaxActive(MAX_ACTIVE_NUMBER);  

  153.   

  154.         //  

  155.         // Creates a connection factory object which will be use by  

  156.         // the pool to create the connection object. We passes the  

  157.         // JDBC url info, username and password.  

  158.         //  

  159.         ConnectionFactory cf = new DriverManagerConnectionFactory(  

  160.                                     CONNECTION_URL,  

  161.                                     USER_NAME,  

  162.                                     PASSWORD);  

  163.   

  164.         //  

  165.         // Creates a PoolableConnectionFactory that will wraps the  

  166.         // connection object created by the ConnectionFactory to add  

  167.         // object pooling functionality.  

  168.         //  

  169.         PoolableConnectionFactory pcf =  

  170.                 new PoolableConnectionFactory(cf, connectionPool,  

  171.                         null, null, false, true);  

  172.         return new PoolingDataSource(connectionPool);  

  173.     }  

  174.   

  175.     public  void printDataSourceStats() {  

  176.         logger.info("Max   : " + getConnectionPool().getMaxActive() + "; " +  

  177.                 "Active: " + getConnectionPool().getNumActive() + "; " +  

  178.                 "Idle  : " + getConnectionPool().getNumIdle());  

  179.     }  

  180.   

  181.     public void shutdownDataSource(DataSource ds) throws SQLException {  

  182.         BasicDataSource bds = (BasicDataSource) ds;  

  183.         bds.close();  

  184.     }  

  185.   

  186.     public String[] getColNames() {  

  187.         return colNames;  

  188.     }  

  189.   

  190.     public void setColNames(String[] colNames) {  

  191.         this.colNames = colNames;  

  192.     }  

  193.   

  194.    

  195. }
复制代码
    SolrIndex.java  实现SolrJ的CRUD操作。
  1. public class SolrIndex {  

  2.   

  3.     Logger logger = LoggerFactory.getLogger(SolrIndex.class) ;  

  4.   

  5.     /*specified the core customer url*/  

  6.     private static final String CORE_CUSTOMER_URL= "http://localhost:8088/solr/customer";  

  7.   

  8.     private  static  HttpSolrServer server;  

  9.     static {  

  10.         server =  new HttpSolrServer(CORE_CUSTOMER_URL);  

  11.         server.setMaxRetries(1); // defaults to 0.  > 1 not recommended.  

  12.         server.setConnectionTimeout(5000); // 5 seconds to establish TCP  

  13.         // Setting the XML response parser is only required for cross  

  14.         // version compatibility and only when one side is 1.4.1 or  

  15.         // earlier and the other side is 3.1 or later.  

  16.         server.setParser(new XMLResponseParser()); // binary parser is used by default  

  17.         // The following settings are provided here for completeness.  

  18.         // They will not normally be required, and should only be used  

  19.         // after consulting javadocs to know whether they are truly required.  

  20.         server.setSoTimeout(1000);  // socket read timeout  

  21.         server.setDefaultMaxConnectionsPerHost(1000);  

  22.         server.setMaxTotalConnections(1000);  

  23.         server.setFollowRedirects(false);  

  24.         // defaults to false  

  25.         // allowCompression defaults to false.  

  26.         // Server side must support gzip or deflate for this to have any effect.  

  27.         server.setAllowCompression(true);  

  28.     }  

  29.     /** 

  30.      * Index a document with specified fields in doc. 

  31.      * @throws IOException 

  32.      * @throws SolrServerException 

  33.      */  

  34.     public void indexDocs() throws IOException, SolrServerException {  

  35.         server.setParser(new XMLResponseParser());  

  36.   

  37.         //Adds the docs and commit them.  

  38.         Collection<SolrInputDocument> docs = new LinkedList<SolrInputDocument>();  

  39.         /* i is used as identification of a document, which is treated as unique key.*/  

  40.         SolrInputDocument doc2 ;  

  41.         /*一千条数据，花费大约一小时，生产660M。使用多线程并发执行估计更好的*/  

  42.         for(int i =10000000; i < 10000002; i++){  

  43.             doc2 = new SolrInputDocument();  

  44.             doc2.addField("customer_id", i);  

  45.             doc2.addField("name", "John Natch-" + i);  

  46.             doc2.addField("level", "VIP");  

  47.             doc2.addField("sex", "男");  

  48.             doc2.addField("address", "【【【【【金刚金刚金刚金刚金刚金】】】】" + i);  

  49.             System.out.println("add doc "+ i);  

  50.             docs.add(doc2);  

  51.             if(docs.size() == 1000){  

  52.                 server.add(docs);  

  53.                 server.commit();  

  54.                 logger.info("commit 1000 doc "+ i);  

  55.                 docs.clear();  

  56.             }  

  57.             /* 

  58.             To immediately commit after adding documents, you could use: 

  59.  

  60.              UpdateRequest req = new UpdateRequest(); 

  61.              req.setAction( UpdateRequest.ACTION.COMMIT, false, false ); 

  62.              req.add( docs ); 

  63.              UpdateResponse rsp = req.process( server ); 

  64.              */  

  65.         }  

  66.         server.add(docs);  

  67.         server.commit();  

  68.         logger.info("Commits successfully!......");  

  69.     }  

  70.   

  71.     /** 

  72.      * solrJ与 database 集成，对数据库中的数据建立索引。当然，这个可以使用Solr DIH取代。 

  73.      */  

  74.     public void indexDocsWithDB(){  

  75.         PoolingDataSourceDemo dataSource = new PoolingDataSourceDemo();  

  76.         List<List<Object>>  rows = dataSource.executeQuerySQL("select * from customer");  

  77.         String[]  columnNames = dataSource.getColNames();  

  78.         Collection<SolrInputDocument> docs = new LinkedList<SolrInputDocument>();  

  79.         SolrInputDocument doc ;  

  80.         for(List row : rows)  {  

  81.             int size = row.size() + 1;  

  82.             doc = new SolrInputDocument();  

  83.             for(int i = 1; i < size ; i++){  

  84.                 doc.addField(columnNames[i], row.get(i-1)) ;  

  85.                 logger.info(columnNames[i]+"add filed "+ row.get(i-1)) ;  

  86.             }  

  87.             docs.add(doc);  

  88.             if(docs.size() > 100){  

  89.                 commitDocs(docs);  

  90.             }  

  91.         }  

  92.         if(docs.size() > 0){  

  93.             commitDocs(docs);  

  94.         }  

  95.     }  

  96.   

  97.     private void commitDocs(Collection<SolrInputDocument> docs){  

  98.         try {  

  99.             //server.deleteById(1)    //specify the id list you want to be deleted.  

  100.             server.add(docs);  

  101.             server.commit();  

  102.             docs.clear();  

  103.         } catch (SolrServerException e) {  

  104.             logger.error("SolrServerException", e);  

  105.         } catch (IOException e) {  

  106.             logger.error("IOException", e) ;  

  107.         }  

  108.     }  

  109.     /** 

  110.      * Query documents with specified query value. 

  111.      * @throws SolrServerException 

  112.      */  

  113.     public void queryDocs() throws SolrServerException {  

  114.         HttpSolrServer server = new HttpSolrServer(CORE_CUSTOMER_URL );  

  115.         server.setParser(new XMLResponseParser());  

  116.   

  117.         /*query  statement settings*/  

  118.         SolrQuery query = new SolrQuery();  

  119.         query.setQuery("李玲");  

  120.         query.setStart(0);  

  121.         query.setRows(10);  

  122.   

  123.         QueryResponse response = server.query( query );  

  124.         SolrDocumentList documents = response.getResults();  

  125.         Iterator<SolrDocument> itr = documents.iterator();  

  126.         logger.info("id   \t   name");  

  127.         while (itr.hasNext()) {  

  128.             SolrDocument doc = itr.next();  

  129.            logger.info(doc.getFieldValue("customer_id") + ":" + "\t"+doc.  

  130.                     getFieldValue("name"));  

  131.         }  

  132.     }  

  133.   

  134.     public void delete(){  

  135.         try {  

  136.             server.deleteByQuery( "*:*" );  

  137.             server.commit();  

  138.         } catch (SolrServerException e) {  

  139.            logger.error("SolrServerException", e);  

  140.         } catch (IOException e) {  

  141.             logger.error("IOException", e);  

  142.         }  

  143.     }  

  144.     public static void main(String[] args){  

  145.         SolrIndex indexer = new SolrIndex();  

  146.         long startTime = System.currentTimeMillis();  

  147.   

  148.         /*do index with specified documents*/  

  149.         try {  

  150.             indexer.indexDocs();  

  151.         } catch (IOException e) {  

  152.             e.printStackTrace();  

  153.         } catch (SolrServerException e) {  

  154.             e.printStackTrace();  

  155.         }  

  156.   

  157. //        try {  

  158. //            indexer.queryDocs();  

  159. //        } catch (SolrServerException e) {  

  160. //            e.printStackTrace();  

  161. //        }  

  162.         /*integration with db. It takes 1214 ms*/  

  163. //        indexer.delete();  

  164. //        indexer.indexDocsWithDB();  

  165.         System.out.println("--------It takes "+ (System.currentTimeMillis() - startTime)  + " ms");  

  166.     }  

  167.   

  168. }
复制代码

      另外，SolrJ操作Solr Cloud的机制与HttpSolrServer一样，除了Http的设置使用CloudSolrServer意外。
  1. <span style="color:#006600;">CloudSolrServer server = new CloudSolrServer("localhost:9983");  

  2. server.setDefaultCollection("collection1");</span>  

  3. SolrInputDocument doc = new SolrInputDocument();  

  4. doc.addField( "id", "1234");  

  5. doc.addField( "name", "A lovely summer holiday");  

  6. server.add(doc);  

  7. server.commit();  
复制代码

          运行代码前，假如下列依赖
1365521106_2133.jpg


欢迎加入about云群371358502、39327136，云计算爱好者群，亦可关注about云腾讯认证空间||关注本站微信
回复

使用道具 举报

自定义广告语

没找到任何评论，期待你打破沉寂

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
gefieder

实习版主

关注

  • 133

    主题

  • 195

    帖子

  • 14

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2025 About云 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000