Product.java
修改Product.java，使得其有更多的字段。 除此之外，还增加了一个toMap()方法，使得其返回为一个Map对象，方便后续的批量增加。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
package
com.how2java;
import
java.util.HashMap;
import
java.util.Map;
public
class
Product {
int
id;
String name;
String category;
float
price;
String place;
String code;
public
int
getId() {
return
id;
}
public
void
setId(
int
id) {
this
.id = id;
}
public
String getName() {
return
name;
}
public
void
setName(String name) {
this
.name = name;
}
public
String getCategory() {
return
category;
}
public
void
setCategory(String category) {
this
.category = category;
}
public
float
getPrice() {
return
price;
}
public
void
setPrice(
float
price) {
this
.price = price;
}
public
String getPlace() {
return
place;
}
public
void
setPlace(String place) {
this
.place = place;
}
public
String getCode() {
return
code;
}
public
void
setCode(String code) {
this
.code = code;
}
@Override
public
String toString() {
return
"Product [id="
+ id +
", name="
+ name +
", category="
+ category +
", price="
+ price +
", place="
+ place +
", code="
+ code +
"]"
;
}
public
Map toMap() {
HashMap<String,Object> map =
new
HashMap<>();
map.put(
"name"
, name);
map.put(
"category"
, category);
map.put(
"code"
, code);
map.put(
"place"
, place);
map.put(
"price"
, price);
return
map;
}
}
|
ProductUtil
准备工具类，把文本文件 140k_products.txt，转换为泛型是Product的集合。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
|
package
com.how2java;
import
java.awt.AWTException;
import
java.io.File;
import
java.io.IOException;
import
java.util.ArrayList;
import
java.util.HashSet;
import
java.util.List;
import
java.util.Set;
import
org.apache.commons.io.FileUtils;
public
class
ProductUtil {
public
static
void
main(String[] args)
throws
IOException, InterruptedException, AWTException {
String fileName =
"140k_products.txt"
;
List<Product> products = file2list(fileName);
System.out.println(products.size());
}
public
static
List<Product> file2list(String fileName)
throws
IOException {
File f =
new
File(fileName);
List<String> lines = FileUtils.readLines(f,
"UTF-8"
);
List<Product> products =
new
ArrayList<>();
for
(String line : lines) {
Product p = line2product(line);
products.add(p);
}
return
products;
}
private
static
Product line2product(String line) {
Product p =
new
Product();
String[] fields = line.split(
","
);
p.setId(Integer.parseInt(fields[
0
]));
p.setName(fields[
1
]);
p.setCategory(fields[
2
]);
p.setPrice(Float.parseFloat(fields[
3
]));
p.setPlace(fields[
4
]);
p.setCode(fields[
5
]);
return
p;
}
}
|
TestElasticSearch4J
通过
ProductUtil 获取到14万条记录的List集合，然后把这个集合通过 BulkRequest 批量地提交到Elastic Search服务器。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
|
package
com.how2java;
import
java.io.IOException;
import
java.util.HashMap;
import
java.util.List;
import
java.util.Map;
import
org.apache.http.HttpHost;
import
org.elasticsearch.ElasticsearchStatusException;
import
org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import
org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import
org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
import
org.elasticsearch.action.bulk.BulkRequest;
import
org.elasticsearch.action.bulk.BulkResponse;
import
org.elasticsearch.action.delete.DeleteRequest;
import
org.elasticsearch.action.get.GetRequest;
import
org.elasticsearch.action.get.GetResponse;
import
org.elasticsearch.action.index.IndexRequest;
import
org.elasticsearch.action.update.UpdateRequest;
import
org.elasticsearch.client.RestClient;
import
org.elasticsearch.client.RestHighLevelClient;
import
org.elasticsearch.common.xcontent.XContentType;
public
class
TestElasticSearch4J {
private
static
RestHighLevelClient client =
new
RestHighLevelClient(
RestClient.builder(
new
HttpHost(
"localhost"
,
9200
,
"http"
)
));
private
static
String indexName =
"how2java"
;
public
static
void
main(String[] args)
throws
IOException {
//确保索引存在
if
(!checkExistIndex(indexName)){
createIndex(indexName);
}
//14万准备数据
List<Product> products = ProductUtil.file2list(
"140k_products.txt"
);
System.out.println(
"准备数据，总计"
+products.size()+
"条"
);
batchInsert(products);
client.close();
}
private
static
void
batchInsert(List<Product> products)
throws
IOException {
// TODO Auto-generated method stub
BulkRequest request =
new
BulkRequest();
for
(Product product : products) {
Map<String,Object> m = product.toMap();
IndexRequest indexRequest=
new
IndexRequest(indexName,
"product"
, String.valueOf(product.getId())).source(m);
request.add(indexRequest);
}
client.bulk(request);
System.out.println(
"批量插入完成"
);
}
private
static
void
deleteDocument(
int
id)
throws
IOException {
DeleteRequest deleteRequest =
new
DeleteRequest (indexName,
"product"
, String.valueOf(id));
client.delete(deleteRequest);
System.out.println(
"已经从ElasticSearch服务器上删除id="
+id+
"的文档"
);
}
private
static
void
updateDocument(Product product)
throws
IOException {
UpdateRequest updateRequest =
new
UpdateRequest (indexName,
"product"
, String.valueOf(product.getId()))
.doc(
"name"
,product.getName());
client.update(updateRequest);
System.out.println(
"已经在ElasticSearch服务器修改产品为："
+product);
}
private
static
void
getDocument(
int
id)
throws
IOException {
// TODO Auto-generated method stub
GetRequest request =
new
GetRequest(
indexName,
"product"
,
String.valueOf(id));
GetResponse response = client.get(request);
if
(!response.isExists()){
System.out.println(
"检查到服务器上 "
+
"id="
+id+
"的文档不存在"
);
}
else
{
String source = response.getSourceAsString();
System.out.print(
"获取到服务器上 "
+
"id="
+id+
"的文档内容是："
);
System.out.println(source);
}
}
private
static
void
addDocument(Product product)
throws
IOException {
Map<String, Object> jsonMap =
new
HashMap<>();
jsonMap.put(
"name"
, product.getName());
IndexRequest indexRequest =
new
IndexRequest(indexName,
"product"
, String.valueOf(product.getId()))
.source(jsonMap);
client.index(indexRequest);
System.out.println(
"已经向ElasticSearch服务器增加产品："
+product);
}
private
static
boolean
checkExistIndex(String indexName)
throws
IOException {
boolean
result =
true
;
try
{
OpenIndexRequest openIndexRequest =
new
OpenIndexRequest(indexName);
client.indices().open(openIndexRequest).isAcknowledged();
}
catch
(ElasticsearchStatusException ex) {
String m =
"Elasticsearch exception [type=index_not_found_exception, reason=no such index]"
;
if
(m.equals(ex.getMessage())) {
result =
false
;
}
}
if
(result)
System.out.println(
"索引:"
+indexName +
" 是存在的"
);
else
System.out.println(
"索引:"
+indexName +
" 不存在"
);
return
result;
}
private
static
void
deleteIndex(String indexName)
throws
IOException {
DeleteIndexRequest request =
new
DeleteIndexRequest(indexName);
client.indices().delete(request);
System.out.println(
"删除了索引："
+indexName);
}
private
static
void
createIndex(String indexName)
throws
IOException {
// TODO Auto-generated method stub
CreateIndexRequest request =
new
CreateIndexRequest(indexName);
client.indices().create(request);
System.out.println(
"创建了索引："
+indexName);
}
}
|
下载地址：http://download.how2j.cn/1703/elasticsearch4j.rar