正文
Elasticsearch Java API简介
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
加入依赖
我本地的Elasticsearch的版本是2.1.0，因此加入相应的maven依赖
<dependency>
<groupId>org.elasticsearch</groupId>
<artifactId>elasticsearch</artifactId>
<version>2.1.0</version>
</dependency>
创建Client
Elasticsearch Client分为Node Client和TransportClient。
- Node Client：节点本身也是Elasticsearch集群的节点，也进入Elasticsearch集群和别的Elasticsearch集群中的节点一样
- TransportClient：轻量级的Client，使用Netty线程池，Socket连接到ES集群。本身不加入到集群，只作为请求的处理
一般我们使用TransportClient。创建Client的实例如下：
private TransportClient client = null;
@Before
public void createElaCLient() throws UnknownHostException {
//如果集群是默认名称的话可以不设置集群名称
Settings settings = Settings.settingsBuilder().put("cluster.name","elasticsearch").build();
client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("master"),9300));
}
/**
* 关闭ela客户端
*/
@After
public void closeElaClient(){
if(client != null){
client.close();
}
}
client.transport.sniff嗅探功能
你可以设置client.transport.sniff为true来使客户端去嗅探整个集群的状态，把集群中其它机器的ip地址加到客户端中，这样做的好处是一般你不用手动设置集群里所有集群的ip到连接客户端，它会自动帮你添加，并且自动发现新加入集群的机器。代码实例如下：
private TransportClient client = null;
@Before
public void createElaCLient() throws UnknownHostException {
//如果集群是默认名称的话可以不设置集群名称
Settings settings = Settings.settingsBuilder().put("cluster.name","elasticsearch").put("client.transport.sniff",true).build();
client = TransportClient.builder().settings(settings).build().addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName("master"),9300));
}
注意：当ES服务器监听使用内网服务器IP而访问使用外网IP时，不要使用client.transport.sniff为true，在自动发现时会使用内网IP进行通信，导致无法连接到ES服务器，而直接使用addTransportAddress方法进行指定ES服务器
测试Client连接到Elasticsearch集群
代码如下：
@Test
public void testConnection(){
List<DiscoveryNode> discoveryList = client.connectedNodes();
for(DiscoveryNode node : discoveryList){
System.out.println(node.getName());
}
}
创建/删除Index和Type信息
/**
* 创建索引
*/
@Test
public void createIndex(){
if(client != null){
client.admin().indices().create(new CreateIndexRequest("test_index")).actionGet();
}
}
/**
* 清除索引
*/
@Test
public void clearIndex(){
IndicesExistsResponse indicesExistsResponse = client.admin().indices().exists(new IndicesExistsRequest("test_index")).actionGet();
if(indicesExistsResponse.isExists()){
client.admin().indices().delete(new DeleteIndexRequest("test_index")).actionGet();
}
}
/**
* 定义索引的映射类型（mapping）
*/
@Test
public void defineIndexTypeMapping(){
try {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject()
.startObject("test")
.startObject("properties")
.startObject("id").field("type","long").field("store","yes").endObject()
.startObject("name").field("type","string").field("store","yes").field("index","not_analyzed").endObject()
.endObject()
.endObject()
.endObject();
PutMappingRequest mappingRequest = Requests.putMappingRequest("test_index").type("test").source(builder);
client.admin().indices().putMapping(mappingRequest).actionGet();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 删除index下的某个type
*/
@Test
public void deleteType(){
if(client != null){
client.prepareDelete().setIndex("test_index").setType("test").execute().actionGet();
}
}
/**
* 创建索引
*/
@Test
public void createIndex(){
if(client != null){
client.admin().indices().create(new CreateIndexRequest("test_index")).actionGet();
}
}
/**
* 清除索引
*/
@Test
public void clearIndex(){
IndicesExistsResponse indicesExistsResponse = client.admin().indices().exists(new IndicesExistsRequest("test_index")).actionGet();
if(indicesExistsResponse.isExists()){
client.admin().indices().delete(new DeleteIndexRequest("test_index")).actionGet();
}
}
/**
* 定义索引的映射类型（mapping）
*/
@Test
public void defineIndexTypeMapping(){
try {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject()
.startObject("test")
.startObject("properties")
.startObject("id").field("type","long").field("store","yes").endObject()
.startObject("name").field("type","string").field("store","yes").field("index","not_analyzed").endObject()
.endObject()
.endObject()
.endObject();
PutMappingRequest mappingRequest = Requests.putMappingRequest("test_index").type("test").source(builder);
client.admin().indices().putMapping(mappingRequest).actionGet();
} catch (IOException e) {
e.printStackTrace();
}
}
/**
* 删除index下的某个type
*/
@Test
public void deleteType(){
if(client != null){
client.prepareDelete().setIndex("test_index").setType("test").execute().actionGet();
}
}
这里自定义了某个Type的索引映射（Mapping），默认ES会自动处理数据类型的映射：针对整型映射为long，浮点数为double，字符串映射为string，时间为date，true或false为boolean。
注意：针对字符串，ES默认会做“analyzed”处理，即先做分词、去掉stop words等处理再index。如果你需要把一个字符串做为整体被索引到，需要把这个字段这样设置：field(“index”, “not_analyzed”)。
索引数据
/**
* 批量索引
*/
@Test
public void indexData(){
BulkRequestBuilder requestBuilder = client.prepareBulk();
for(Person person : personList){
String obj = getIndexDataFromHotspotData(person);
if(obj != null){
requestBuilder.add(client.prepareIndex("test_index","test",String.valueOf(person.getId())).setRefresh(true).setSource(obj));
}
}
BulkResponse bulkResponse = requestBuilder.execute().actionGet();
if(bulkResponse.hasFailures()){
Iterator<BulkItemResponse> it = bulkResponse.iterator();
while(it.hasNext()){
BulkItemResponse itemResponse = it.next();
if(itemResponse.isFailed()){
System.out.println(itemResponse.getFailureMessage());
}
}
}
}
/**
* 单个索引数据
* @return
*/
@Test
public void indexHotspotData() {
String jsonSource = getIndexDataFromHotspotData(new Person(1004,"jim"));
if (jsonSource != null) {
IndexRequestBuilder requestBuilder = client.prepareIndex("test_index",
"test").setRefresh(true);
requestBuilder.setSource(jsonSource)
.execute().actionGet();
}
}
public String getIndexDataFromHotspotData(Person p){
String result = null;
if(p != null){
try {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject().field("id",p.getId()).field("name",p.getName()).endObject();
result = builder.string();
} catch (IOException e) {
e.printStackTrace();
}
}
return result;
}
查询数据
/**
* 批量索引
*/
@Test
public void indexData(){
BulkRequestBuilder requestBuilder = client.prepareBulk();
for(Person person : personList){
String obj = getIndexDataFromHotspotData(person);
if(obj != null){
requestBuilder.add(client.prepareIndex("test_index","test",String.valueOf(person.getId())).setRefresh(true).setSource(obj));
}
}
BulkResponse bulkResponse = requestBuilder.execute().actionGet();
if(bulkResponse.hasFailures()){
Iterator<BulkItemResponse> it = bulkResponse.iterator();
while(it.hasNext()){
BulkItemResponse itemResponse = it.next();
if(itemResponse.isFailed()){
System.out.println(itemResponse.getFailureMessage());
}
}
}
}
/**
* 单个索引数据
* @return
*/
@Test
public void indexHotspotData() {
String jsonSource = getIndexDataFromHotspotData(new Person(1004,"jim"));
if (jsonSource != null) {
IndexRequestBuilder requestBuilder = client.prepareIndex("test_index",
"test").setRefresh(true);
requestBuilder.setSource(jsonSource)
.execute().actionGet();
}
}
public String getIndexDataFromHotspotData(Person p){
String result = null;
if(p != null){
try {
XContentBuilder builder = XContentFactory.jsonBuilder();
builder.startObject().field("id",p.getId()).field("name",p.getName()).endObject();
result = builder.string();
} catch (IOException e) {
e.printStackTrace();
}
}
return result;
}
ES支持分页查询获取数据，也可以一次性获取大量数据，需要使用Scroll Search，QueryBuilder是一个查询条件
public List<Long> searchData(QueryBuilder builder){
List<Long> ids = new ArrayList<>();
SearchResponse response = client.prepareSearch("test_index").setTypes("test").setQuery(builder).setSize(10).execute().actionGet();
SearchHits hits = response.getHits();
for(SearchHit hit : hits){
Long id = (Long) hit.getSource().get("id");
ids.add(id);
}
return ids;
}
