SolrJ 操作HttpSolrServer, ConcurrentUpdateSolrServer和CloudSolrServer

    HttpSolrServer 使用了Apache Commons HTTP 客户端来连接Solr. 注意在Solr 4.x中， CommonsHttpSolrServer 已经改变为HttpSolrServer 以及StreamingUpdateSolrServer 已经改变为ConcurrentUpdateSolrServer  。ConcurrentUpdateSolrServer 更适合update 操作，而HttpSolrServer 更适合query操作。

       添加document或是修改document。假如这个document已经存在，就会update这个document。代码片段如下:

[java] view plaincopyprint?

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中配置具体的字段。

[html] view plaincopyprint?

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"/>  

[html] view plaincopyprint?

1. <uniqueKey>customer_id</uniqueKey>  

 

        删除操作：

[html] view plaincopyprint?

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实现。两种各有其优缺点，根据实际的应用来选择具体的实现方式。

  

[html] view plaincopyprint?

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   实现线程池连接数据库。

[java] view plaincopyprint?

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操作。

[java] view plaincopyprint?

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意外。

[java] view plaincopyprint?

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();  

          运行代码前，假如下列依赖