Java操作ElasticSearch
 
1.Java连接ElasticSearch
 
1）创建普通maven工程，导入依赖
 
 
2）测试连接ElasticSearch
 

 
 
2.Java创建索引
 

 
 
 
 
3.Java查询删除索引
 

 
 
 
 
 
4.Java操作文档（对文档的增删改，批量增删）
 
1）创建person类
 
 
2）创建生成文档
 

 
 
3）修改文档
 

 
 
4）删除文档
 

 
 
5）批量生成文档
 

 
 
6）批量删除文档
 

 

实体类：利用注解建立实体类会自动根据实体类创建索引，不需要自己建立mapping
 
package com.msic.mall.search.domain;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.Document;
import org.springframework.data.elasticsearch.annotations.Field;
import org.springframework.data.elasticsearch.annotations.FieldType;

import java.io.Serializable;
import java.math.BigDecimal;
import java.util.List;

/**
 * 搜索中的商品信息
 * Created by macro on 2018/6/19.
 */
@Data
@NoArgsConstructor
@AllArgsConstructor
@Accessors(chain = true)
@Document(indexName = "pms_product", type = "_doc",shards = 1,replicas = 0)
public class EsProduct implements Serializable {

    @Id
    private Long id;
    private Long shopId;
    @Field(type = FieldType.Keyword)
    private String shopName;
    private String shopIcon;

    private Long spuId;

    @Field(analyzer = "ik_max_word",type = FieldType.Text)
    private String productName;
    private String productIcon;

    private BigDecimal productPrice;


    @Field(type = FieldType.Integer)
    private Integer sales; //销量

    private Integer stock;//库存,
    private boolean hasStock;//库存,

    @Field(type = FieldType.Integer)
    private  Integer newStatus; //新品

    @Field(type = FieldType.Integer)
    private  Integer advanceSaleStatus;//预售


    private Long brandId;//    品牌id
    @Field(type = FieldType.Keyword)
    private String brandName;//    品牌名称
    private Long productCategoryId;  //分类id
    @Field(type = FieldType.Keyword)
    private String productCategoryName;  //商品分类名称

//    @Field(analyzer = "ik_max_word",type = FieldType.Text)
//    private String subTitle; //    副标题
//    @Field(analyzer = "ik_max_word",type = FieldType.Text)
//    private String keywords;
    private Integer sort;   //排序

//    @Field(type = FieldType.Keyword)
    private String specificationList;

    private  Long hotScore;//    热度值

//    @Field(type = FieldType.Nested)
//    private List<EsProductAttributeValue> attrValueList;

    private static final long serialVersionUID = -1L;


//    货号
//    @Field(type = FieldType.Keyword)
//    private String productSn;

//    private Integer promotionType;
//    是否推荐
//    private Integer recommandStatus;

}
 
controller
 

    @ApiOperation(value = "导入所有数据库中商品到ES")
    @PostMapping(value = "/importAll")
    public CommonResponseResult<Integer> importAllList() {
        CommonResponseResult responseResult = null;
        Map map = new HashMap();
        try {
            int count = esProductService.importAll();
            map.put("result",count+"条记录被添加");
            responseResult=CommonResponseResult.ok(map);
        }catch (Exception e){
            e.printStackTrace();
            responseResult=CommonResponseResult.notOk(e.getMessage());
        }
        return responseResult;
    }
 
service：查出需要导入的数据
 
 @Override
    public int importAll() {
        List<EsProduct> esProductList = productDao.getAllEsProductList(null);//从数据库查询数据
        for (EsProduct product:
             esProductList) {
            product.setProductIcon(dfsUrlPrefix+product.getProductIcon());
            LOGGER.info(product.getSpecificationList());
        }
        int result = 0;
        try {
            try {
                productRepository.deleteAll();
            }catch (Exception e){
                e.printStackTrace();
                throw new MyException("原数据删除失败");
            }
            Iterable<EsProduct> esProductIterable = productRepository.saveAll(esProductList);
            Iterator<EsProduct> iterator = esProductIterable.iterator();
            while (iterator.hasNext()) {
                result++;
                iterator.next();
            }
        }catch (Exception e){
            e.printStackTrace();
            throw new MyException("数据导入失败，请重试");
        }
        return result;
    }
 
 
 
 
 
 

 
前言
 
本文介绍了如何整合搜索引擎elasticsearch与springboot，对外提供数据查询接口。
 
业务介绍
 
我的个人网站需要对mysql数据库内存储的京东商品进行模糊查询（模仿淘宝商品搜索），所以选择了将数据导入elasticsearch随后使用他来进行关键词查询。前端只需发送用户搜索的关键词和分页参数（可选），即可返回商品数据（json格式）
 
开发环境
 
组件介绍：
 
elasticsearch：搜索引擎，用于存储待搜索数据
logstash：用于将mysql中的商品数据同步到搜索引擎中
elasticsearch-head（可选）：elasticsearch可视化工具
kibana（可选）：elasticsearch可视化工具
 
本文测试环境：
 
springboot：1.5.16
elasticsearch：2.3.5（springboot1.5仅支持2.x的es）
logstash：6.5.4
 
开发步骤
 
使用Docker部署elasticsearch
 
docker下一键启动es，可根据需要的版本号对语句做修改
 
sudo docker run -it --rm --name elasticsearch -d -p 9200:9200 -p 9300:9300 elasticsearch:2.3.5
 
注意到该命令：
 
–rm参数：容器终止后销毁
-d：后台进程
-p 9200:9200 -p 9300:9300：开放了9200端口和9300端口
 
得到如图：
 
 
此时打开网页localhost:9200即可查看状态，显示类似为：
 
{
  "name" : "Ant-Man",
  "cluster_name" : "elasticsearch",
  "version" : {
    "number" : "2.3.5",
    "build_hash" : "90f439ff60a3c0f497f91663701e64ccd01edbb4",
    "build_timestamp" : "2016-07-27T10:36:52Z",
    "build_snapshot" : false,
    "lucene_version" : "5.5.0"
  },
  "tagline" : "You Know, for Search"
}
 
注意：docker的es默认对0.0.0.0公网开放
 
下载并使用logstash并导入数据
 
本文中要导入的是pm_backend下的表pm_jd_item内的全部京东商品数据
 
详细步骤参考：
 
http://blog.codecp.org/2018/04/16/Elasticsearch%E4%B9%8B%E4%BD%BF%E7%94%A8Logstash%E5%AF%BC%E5%85%A5Mysql%E6%95%B0%E6%8D%AE/
 
最终编写的jdbc.conf为：
 
schedule => "* * * * *"默认为每分钟同步一次
 
input {
  jdbc {
    jdbc_connection_string => "jdbc:mysql://localhost:3306/pm_backend"
    jdbc_user => "root"
    jdbc_password => "xxxxxxxxxx"
    jdbc_driver_library => "xxxxxxxx/mysql-connector-java-5.1.6.jar"
    jdbc_driver_class => "com.mysql.jdbc.Driver"
    jdbc_paging_enabled => "true"
    jdbc_page_size => "5000"
    statement=> "select * from pm_jd_item"
    schedule => "* * * * *"
    type => "pm_jd_item"
  }
}

output {
  elasticsearch {
    hosts => "localhost:9200"
    index => "pm_backend"
    document_type => "%{type}"
    document_id => "%{id}"
  }
  stdout {
    codec => json_lines
  }
}
 
在logstash目录下执行命令，完成数据的导入：
 
bin/logstash -f jdbc.conf
 
得到如图：
 
 
同步完成后，使用elasticsearch-head查看（或者用kibana，请随意）：
 
 
整合进springboot
 
添加pom.xml
 
<!-- 搜索引擎：elastic-search-->
<dependency>
	<groupId>org.elasticsearch</groupId>
	<artifactId>elasticsearch</artifactId>
	 <version>2.4.6</version>
</dependency>
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>
<dependency>
	<groupId>org.springframework.data</groupId>
	<artifactId>spring-data-elasticsearch</artifactId>
</dependency>
 
修改application.properties
 
# elasticsearch
spring.data.elasticsearch.cluster-name=elasticsearch
#节点地址，多个节点用逗号隔开
spring.data.elasticsearch.cluster-nodes=127.0.0.1:9300
#spring.data.elasticsearch.local=false
spring.data.elasticsearch.repositories.enable=true
 
在需要进行搜索的实体类上添加@Document、@Id、@Field等标注，本例为JdItem.java
 
@Document(indexName = "pm_backend", type = "pm_jd_item")
public class JdItem implements Serializable {

    @Id
    private Integer id;

    @Field(type = FieldType.Long)
    private Long itemId;

    @Field(type = FieldType.Long)
    private Long categoryId;

    @Field(type = FieldType.String)
    private String name;
 
添加JdItemRepository继承ElasticsearchRepository
 
public interface JdItemRepository extends ElasticsearchRepository<JdItem, Integer>{
}

 
编写JdItemController中的查询接口findJdItemByName
 
代码截取自个人项目京东价格监控，仅供参考！
 
    /**
     * 根据商品名在pm_jd_item中搜索商品
     * @param itemName
     * @param startRow
     * @param pageSize
     * @return
     */
    @ApiOperation(value="查询商品", notes="查询商品")
    @RequestMapping(value = "/findJdItemByName", method = {RequestMethod.GET})
    public ResponseData<List<JdItem>> findJdItemByName(
            @ApiParam("用户输入的商品名") @RequestParam(value = "itemName") String itemName,
            @ApiParam("页码索引(默认为0)") @RequestParam(value = "startRow", required = false, defaultValue = "0") int startRow,
            @ApiParam("每页的商品数量(默认为10)") @RequestParam(value = "pageSize", required = false, defaultValue = "10") int pageSize
    ){
        ResponseData<List<JdItem>> responseData = new ResponseData<>();
        try {

            FunctionScoreQueryBuilder functionScoreQueryBuilder = QueryBuilders.functionScoreQuery().add(QueryBuilders.matchPhraseQuery("name", itemName), ScoreFunctionBuilders.weightFactorFunction(100)).scoreMode("sum").setMinScore(10);
            Pageable pageable = new PageRequest(startRow, pageSize);
            SearchQuery searchQuery = new NativeSearchQueryBuilder().withPageable(pageable).withQuery(functionScoreQueryBuilder).build();
            Page<JdItem> jdItems = jdItemRepository.search(searchQuery);
            // Page分页getTotalPages()返回了应有的页数，临时放在errorMsg传给前端
            responseData.jsonFill(1, String.valueOf(jdItems.getTotalPages()), jdItems.getContent());
        } catch (Exception e) {
            e.printStackTrace();
            responseData.jsonFill(2, e.getMessage(), null);
        }
        return responseData;
    }
}
 
运行springboot
 
调用findJdItemByName接口，得到：
 
 
整合分词器功能
 
请参考：https://github.com/medcl/elasticsearch-analysis-ik
 
参考
 
Docker安装ES & Kibana：
 
https://www.jianshu.com/p/fdfead5acc23
 
Elasticsearch之使用Logstash导入Mysql数据：
 
http://blog.codecp.org/2018/04/16/Elasticsearch%E4%B9%8B%E4%BD%BF%E7%94%A8Logstash%E5%AF%BC%E5%85%A5Mysql%E6%95%B0%E6%8D%AE/
 
关注我
 
我是蛮三刀把刀，后端开发。主要关注后端开发，数据安全，爬虫等方向。
 
来微信和我聊聊：yangzd1102
 
Github个人主页：
 
https://github.com/qqxx6661
 
原创博客主要内容
 
Java知识点复习全手册
Leetcode算法题解析
剑指offer算法题解析
Python爬虫相关技术实战
后端开发相关技术实战
SpringCloud实战
 
同步更新公众号及以下全部博客：
 
1. Csdn
 
http://blog.csdn.net/qqxx6661
 
2. 知乎
 
https://www.zhihu.com/people/yang-zhen-dong-1/
 
3. 掘金
 
https://juejin.im/user/5b48015ce51d45191462ba55
 
4. 简书
 
https://www.jianshu.com/u/b5f225ca2376
 
个人公众号：Rude3Knife
 
 
如果文章对你有帮助，不妨收藏起来并转发给您的朋友们~

上次写了一个Java整合Storm的一个最简单的例子https://blog.csdn.net/xxkalychen/article/details/117018310，一个数据源，一个处理中间环节，一个最终环节，一个任务拓扑。我们能看到的效果是控制台打印。根据需要，我们要把数据写入ElasticSearch。
 
我们在原有的项目中做一点小的调整，来实现这个需求。
 
当然，首先我们需要一个ElasticSearch的服务器。我已经开启了自己架设好的ES服务器，版本号7.12.0
 
 
一、pom中添加ElasticSearch和Gson依赖
 
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
    <version>7.12.0</version>
</dependency>
<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.8.6</version>
</dependency>
 
这是ElasticSearch7.x以上版本官方建议使用的包，大版本要和服务器一致。Gson用于将对象构建成json，因为ElasticSearch需要将数据构建为合法的json进行存储。
 
二、创建一个数据包装类WordPkg。由于我们处理的最终数据都是单词，是字符串。我们需要一个类，将单词包装为有字段识别的类。
 
package com.chris.storm.model;

import java.io.Serializable;

/**
 * @author Chris Chan
 * Create on 2021/5/19 10:37
 * Use for:
 * Explain:
 */
public class WordPkg implements Serializable {
    private String word;

    public WordPkg() {
    }

    public WordPkg(String word) {
        this.word = word;
    }

    public String getWord() {
        return word;
    }

    public void setWord(String word) {
        this.word = word;
    }
}
 
三、为使用方便，我写了一个ElasticSearchUtil工具类，来简化操作
 
package com.chris.storm.utils;

import com.chris.storm.model.WordPkg;
import com.google.gson.Gson;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;

import java.io.IOException;
import java.util.UUID;

/**
 * @author Chris Chan
 * Create on 2021/5/19 7:37
 * Use for:
 * Explain:
 */
public class ElasticSearchUtil {
    private static RestHighLevelClient client = null;

    static {
        ElasticSearchUtil.client = new RestHighLevelClient(RestClient.builder(new HttpHost("192.168.0.52", 9200, "http")));
    }

    public static RestHighLevelClient getClient() {
        return client;
    }

    public static void close() {
        try {
            ElasticSearchUtil.client.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static boolean isIndexExists(String indexName) {
        try {
            return ElasticSearchUtil.client.indices().exists(new GetIndexRequest(indexName), RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    public static void createIndex(String indexName) {
        if (isIndexExists(indexName)) {
            return;
        }
        try {
            ElasticSearchUtil.client.indices().create(new CreateIndexRequest(indexName), RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static IndexResponse add(WordPkg wordPkg, String indexName) {
        IndexRequest indexRequest = new IndexRequest(indexName).id(UUID.randomUUID().toString());
        indexRequest.source(new Gson().toJson(wordPkg), XContentType.JSON);

        try {
            return ElasticSearchUtil.client.index(indexRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

}
 
四、修改PrintBolt。我们需要添加写入ElasticSearch的逻辑。
 
package com.chris.storm.bolt;

import com.chris.storm.model.WordPkg;
import com.chris.storm.utils.ElasticSearchUtil;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

/**
 * @author Chris Chan
 * Create on 2021/5/19 9:44
 * Use for:
 * Explain:
 */
public class PrintBolt extends BaseRichBolt {
    //在ElasticSearch中创建的索引名称
    public static final String INDEX_NAME = "storm_word";

    static {
        ElasticSearchUtil.createIndex(INDEX_NAME);
    }

    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        
    }

    @Override
    public void execute(Tuple tuple) {
        String word = tuple.getStringByField("word");
        System.out.printf("%s: %s\n", "接收到单词", word);
        //输出到ElasticSearch
        ElasticSearchUtil.add(new WordPkg(word), INDEX_NAME);

    }

    /**
     * 这是流水线上的终点，不需要在发给下一环，所以无须再定义元组字段
     *
     * @param outputFieldsDeclarer
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }
}
 
因为封装了工具类，所以使用看起来就特别简单。
 
除了这一处，其他的地方都不需要修改。直接运行测试即可。
 
五、运行测试
 
 
 
这个程序还是不能远程提交，我们还需要进行修改。