yangyoupeng / zhaoonline-utils

Go to a project
Toggle navigation Toggle navigation pinning
yangyoupeng
e4e32e34 1 parent 3abc1025

添加bulkAdd方法以及测试例子

Inline Side-by-side
Showing 4 changed files with 215 additions and 0 deletions
...@@ -8,6 +8,7 @@ import java.util.Map; ...@@ -8,6 +8,7 @@ import java.util.Map;
8 import org.apache.http.HttpEntity; 8 import org.apache.http.HttpEntity;
9 import org.apache.http.HttpHost; 9 import org.apache.http.HttpHost;
10 import org.apache.http.HttpRequest; 10 import org.apache.http.HttpRequest;
11 +import org.apache.http.ParseException;
11 import org.apache.http.client.methods.CloseableHttpResponse; 12 import org.apache.http.client.methods.CloseableHttpResponse;
12 import org.apache.http.client.methods.RequestBuilder; 13 import org.apache.http.client.methods.RequestBuilder;
13 import org.apache.http.entity.ContentType; 14 import org.apache.http.entity.ContentType;
...@@ -16,8 +17,10 @@ import org.apache.http.impl.client.CloseableHttpClient; ...@@ -16,8 +17,10 @@ import org.apache.http.impl.client.CloseableHttpClient;
16 import org.apache.http.util.EntityUtils; 17 import org.apache.http.util.EntityUtils;
17 18
18 import com.fasterxml.jackson.core.JsonParseException; 19 import com.fasterxml.jackson.core.JsonParseException;
20 +import com.fasterxml.jackson.core.JsonProcessingException;
19 import com.fasterxml.jackson.databind.JsonMappingException; 21 import com.fasterxml.jackson.databind.JsonMappingException;
20 import com.zhaoonline.common.Utils.IOUtils; 22 import com.zhaoonline.common.Utils.IOUtils;
23 +import com.zhaoonline.common.es.bean.BulkResponse;
21 import com.zhaoonline.common.es.bean.DeleteResponse; 24 import com.zhaoonline.common.es.bean.DeleteResponse;
22 import com.zhaoonline.common.es.bean.GetResponse; 25 import com.zhaoonline.common.es.bean.GetResponse;
23 import com.zhaoonline.common.es.bean.IndexResponse; 26 import com.zhaoonline.common.es.bean.IndexResponse;
...@@ -277,4 +280,94 @@ public class ESHttpClient { ...@@ -277,4 +280,94 @@ public class ESHttpClient {
277 IOUtils.closeStream(response); 280 IOUtils.closeStream(response);
278 return deleteResponse.isSuccess(); 281 return deleteResponse.isSuccess();
279 } 282 }
283 +
284 + /**
285 + * Method name: bulkAddDoc <BR>
286 + * Description: 批量添加指定ID <BR>
287 + * Remark: <BR>
288 + * @param ids
289 + * @param dataList
290 + * @return
291 + * @throws UnsupportedOperationException
292 + * @throws IOException boolean<BR>
293 + */
294 + public boolean bulkAddDoc(List ids,List dataList) throws UnsupportedOperationException, IOException {
295 + if(!validateBulkInput(ids, dataList)){
296 + throw new IllegalArgumentException("the ids size is not equal to dataList ");
297 + };
298 +
299 + StringBuilder builder=buildBulkIndexScript(ids,dataList);
300 + if(builder.length()==0){
301 + return false;
302 + }
303 + return bulkAdd(builder.toString());
304 + }
305 +
306 + private boolean validateBulkInput(List ids, List dataList) {
307 + if(ids !=null || dataList != null){
308 + return false;
309 + }
310 + if(ids.size() != dataList.size() ){
311 + return false;
312 + }
313 + return true;
314 + }
315 +
316 +
317 + /**
318 + * Method name: bulkAddDoc <BR>
319 + * Description: 批量添加自动生成ID <BR>
320 + * Remark: <BR>
321 + * @param dataList
322 + * @return
323 + * @throws UnsupportedOperationException
324 + * @throws IOException boolean<BR>
325 + */
326 + public boolean bulkAddDoc(List dataList) throws UnsupportedOperationException, IOException {
327 + StringBuilder builder=buildBulkIndexScript(null,dataList);
328 + if(builder.length()==0){
329 + return false;
330 + }
331 + return bulkAdd(builder.toString());
332 + }
333 +
334 + public boolean bulkAdd(String bulkScripts) throws ParseException, IOException{
335 + String bulkPath=path+PATH_SEPERATOR+"_bulk";
336 + CloseableHttpResponse response= sendPostRequest(bulkScripts,bulkPath,this.hostsList);
337 + if(response ==null){
338 + return false;
339 + }
340 + HttpEntity responseEntity=response.getEntity();
341 + BulkResponse bulkResponse=JsonUtils.toObject(responseEntity.getContent(), BulkResponse.class);
342 + IOUtils.closeStream(response);
343 + return !bulkResponse.isErrors();
344 + }
345 +
346 +
347 + public StringBuilder buildBulkIndexScript(List ids,List dataList) throws JsonProcessingException {
348 + StringBuilder bulkBuilder=new StringBuilder();
349 + if(ids !=null && ids.size() !=0){
350 + for(int i=0;i<ids.size();i++){
351 + StringBuilder itemBuilder=new StringBuilder();
352 + itemBuilder.append("{ \"index\" :").append("{ \"_index\" : \"").
353 + append(this._index).append("\"").append(",\"_type\" :\"").append(this._type).append("\"")
354 + .append(",\"_id\" :\"").append(ids.get(i)).append("\"")
355 + .append("} }").append("\n");
356 + String objectjson=JsonUtils.toJson(dataList.get(i));
357 + itemBuilder.append(objectjson).append("\n");
358 + bulkBuilder.append(itemBuilder);
359 + }
360 + }else {
361 + for(Object object:dataList){
362 + StringBuilder itemBuilder=new StringBuilder();
363 + itemBuilder.append("{ \"create\" :").append("{ \"_index\" : \"").
364 + append(this._index).append("\"").append(",\"_type\" :\"").append(this._type).append("\"")
365 + //.append(",\"_id\" :\"").append(1).append("\"")
366 + .append("} }").append("\n");
367 + String objectjson=JsonUtils.toJson(object);
368 + bulkBuilder.append(itemBuilder);
369 + }
370 + }
371 + return bulkBuilder;
372 + }
280 } 373 }
......
1 +package com.zhaoonline.common.es.bean;
2 +
3 +import java.util.List;
4 +
5 +import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
6 +
7 +@JsonIgnoreProperties(ignoreUnknown = true)
8 +public class BulkResponse {
9 + private long took;
10 + private boolean errors;
11 + private List items;
12 + public long getTook() {
13 + return took;
14 + }
15 + public void setTook(long took) {
16 + this.took = took;
17 + }
18 + public boolean isErrors() {
19 + return errors;
20 + }
21 + public void setErrors(boolean errors) {
22 + this.errors = errors;
23 + }
24 + public List getItems() {
25 + return items;
26 + }
27 + public void setItems(List items) {
28 + this.items = items;
29 + }
30 +
31 +}
1 package com.zhaoonline.common.es.bean; 1 package com.zhaoonline.common.es.bean;
2 2
3 import com.fasterxml.jackson.annotation.JsonIgnore; 3 import com.fasterxml.jackson.annotation.JsonIgnore;
4 +import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
4 import com.fasterxml.jackson.annotation.JsonProperty; 5 import com.fasterxml.jackson.annotation.JsonProperty;
5 6
7 +@JsonIgnoreProperties(ignoreUnknown = true)
6 public class IndexResponse extends CommonESResponse{ 8 public class IndexResponse extends CommonESResponse{
7 9
8 @JsonProperty("created") 10 @JsonProperty("created")
......
1 package com.zhaoonline.common.es; 1 package com.zhaoonline.common.es;
2 2
3 import java.io.IOException; 3 import java.io.IOException;
4 +import java.util.ArrayList;
4 import java.util.HashMap; 5 import java.util.HashMap;
5 import java.util.HashSet; 6 import java.util.HashSet;
7 +import java.util.List;
6 import java.util.Map; 8 import java.util.Map;
7 import java.util.Set; 9 import java.util.Set;
8 10
...@@ -160,6 +162,93 @@ public class TestESClient { ...@@ -160,6 +162,93 @@ public class TestESClient {
160 client.close(); 162 client.close();
161 } 163 }
162 164
165 + @Test
166 + public void testBuilderBulkScript() throws JsonProcessingException{
167 + String index="zhaoon1";
168 + String type="test1";
169 + ESClientConfiguration config=new ESClientConfiguration();
170 + config.addHostPorts("127.0.0.1:9200");
171 + ESHttpClient client= new ESHttpClient(index,type,config);
172 +
173 + List dataList=new ArrayList<>();
174 + for(int i=0;i<1;i++){
175 + Map object=new HashMap<>();
176 + object.put("testKey"+i, "testValue");
177 + dataList.add(object);
178 + }
179 + StringBuilder string=client.buildBulkIndexScript(null,dataList);
180 +
181 + //
182 + StringBuilder expectBuilder=new StringBuilder();
183 + expectBuilder.append("{ \"create\" :").append("{ \"_index\" : \"").
184 + append("zhaoon1").append("\"").append(",\"_type\" :\"").append("test1").append("\"").append("} }").append("\n");
185 + Map object=new HashMap<>();
186 + object.put("testKey0", "testValue");
187 + String objectjson=JsonUtils.toJson(object);
188 + expectBuilder.append(objectjson).append("\n");
189 +
190 + Assert.assertEquals(expectBuilder.toString(), string.toString());
191 + }
163 192
193 + @Test
194 + public void testBulkAddAutoId() throws UnsupportedOperationException, IOException{
195 +
196 + String index="zhaoon1";
197 + String type="test1";
198 + ESClientConfiguration config=new ESClientConfiguration();
199 + config.addHostPorts("127.0.0.1:9200");
200 + ESHttpClient client= new ESHttpClient(index,type,config);
201 + client.init();
202 +
203 + List dataList=new ArrayList<>();
204 + for(int i=0;i<100;i++){
205 + Map object=new HashMap<>();
206 + object.put("testKey"+i, "testValue");
207 + dataList.add(object);
208 + }
209 + boolean result=client.bulkAddDoc(dataList);
210 + Assert.assertTrue(result);
211 + }
212 +
213 + @Test(expected=IllegalArgumentException.class)
214 + public void testBulkAddIdsNotMatchDataList() throws UnsupportedOperationException, IOException{
215 +
216 + String index="zhaoon1";
217 + String type="test1";
218 + ESClientConfiguration config=new ESClientConfiguration();
219 + config.addHostPorts("127.0.0.1:9200");
220 + ESHttpClient client= new ESHttpClient(index,type,config);
221 + client.init();
222 + List ids=new ArrayList<>();
223 + List dataList=new ArrayList<>();
224 + for(int i=0;i<100;i++){
225 + Map object=new HashMap<>();
226 + object.put("testKey"+i, "testValue");
227 + dataList.add(object);
228 + }
229 + boolean result=client.bulkAddDoc(ids,dataList);
230 + }
231 +
232 +
233 + @Test
234 + public void testBulkAdd() throws UnsupportedOperationException, IOException{
235 +
236 + String index="zhaoon1";
237 + String type="test1";
238 + ESClientConfiguration config=new ESClientConfiguration();
239 + config.addHostPorts("127.0.0.1:9200");
240 + ESHttpClient client= new ESHttpClient(index,type,config);
241 + client.init();
242 + List ids=new ArrayList<>();
243 + List dataList=new ArrayList<>();
244 + for(int i=0;i<100;i++){
245 + Map object=new HashMap<>();
246 + object.put("testKey"+i, "testValue");
247 + dataList.add(object);
248 + ids.add(i);
249 + }
250 + boolean result=client.bulkAddDoc(ids,dataList);
251 + Assert.assertTrue(result);
252 + }
164 253
165 } 254 }
......