第六章、Elasticsearch 分布式搜索引擎

1、Elasticsearch入门

下载与安装

下载的版本最好与SpringBoot中一致

安装ES

https://www.elastic.co/cn/downloads/past-releases/elasticsearch-6-4-3

配置

elasticsearch.yml

cluster.name: my-application # 集群名称

path.data: D:\attachment\elasticsearch-6.4.3\data

path.logs: D:\attachment\elasticsearch-6.4.3\data\logs

配置高级系统设置path

安装分词器

默认支持英文分词，我们需要安装中文分词器

下载地址：https://github.com/medcl/elasticsearch-analysis-ik

下载完毕后直接解压，必须解压到ES的plugin文件下的ik文件下

启动Elasticsearch

查看集群健康状况

curl -X GET "localhost:9200/_cat/health?v

查看节点

curl -X GET "localhost:9200/_cat/nodes?v

查看索引

curl -X GET "localhost:9200/_cat/indices?v

建立索引

curl -X PUT "localhost:9200/test"

{"acknowledged":true,"shards_acknowledged":true,"index":"test"} # 成功

删除索引

curl -X DELETE "localhost:9200/test"
{"acknowledged":true}

可以利用postman执行语句

查询所有

localhost:9200/test/_search

根据id查询

localhost:9200/test/_doc/1

根据id删除

localhost:9200/test/_doc/1

添加数据

多条件查询

2、Spring 整合Elasticsearch

1. 引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

2. 配置Es

# elasticsearch
spring.data.elasticsearch.cluster-name=nowcoder
spring.data.elasticsearch.cluster-nodes=127.0.0.1:9300

3. netty启动冲突问题

@SpringBootApplication
public class CommunityApplication {

    @PostConstruct//构造器调用完后被执行
    public void init(){
        // 解决netty启动冲突问题
        System.setProperty("es.set.netty.runtime.available.processors","false");
    }


    public static void main(String[] args) {
        SpringApplication.run(CommunityApplication.class, args);
    }

}

配置实体类

@Document(indexName = "discusspost",type = "_doc",shards = 6,replicas = 3)
public class DiscussPost {
    @Field(type = FieldType.Integer)
    private Integer id;
    @Field(type = FieldType.Integer)
    private Integer userId;
    @Field(type = FieldType.Text, analyzer = "ik_max_word", searchAnalyzer = "ik_smart")
    private String title;
    @Field(type = FieldType.Text, analyzer = "ik_max_word", searchAnalyzer = "ik_smart")
    private String content;
    // 类型 0：普通 1：置顶
    @Field(type = FieldType.Integer)
    private Integer type;
    // 状态 0：正常 1：精华 2：拉黑
    @Field(type = FieldType.Integer)
    private Integer status;
    @Field(type = FieldType.Date)
    private Date createTime;
    @Field(type = FieldType.Integer)
    private Integer commentCount; // 评论数量
    @Field(type = FieldType.Double)
    private double score;
}

编写接口

DiscussPostRepository

@Repository
public interface DiscussPostRepository extends ElasticsearchRepository<DiscussPost,Integer> {

}

测试ES

@SpringBootTest
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = CommunityApplication.class)
public class ElasticsearchTest {

    @Autowired
    private DiscussPostMapper discussPostMapper;

    @Autowired
    private DiscussPostRepository discussPostRepository;

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    /**
     * 插入一条
     */
    @Test
    public void testInsert() {
        // 向es中存入数据
        discussPostRepository.save(discussPostMapper.selectDiscussPostById(19));
        discussPostRepository.save(discussPostMapper.selectDiscussPostById(31));
        discussPostRepository.save(discussPostMapper.selectDiscussPostById(36));

    }

    /**
     * 插入多条
     */
    @Test
    public void testInsertList() {
        discussPostRepository.saveAll(discussPostMapper.selectDiscussPosts(111, 0, 5));
    }

    /**
     * 修改数据
     */
    @Test
    public void testUpdate() {
        DiscussPost discussPost = discussPostMapper.selectDiscussPostById(19);
        discussPost.setContent("我是新人，使劲关注");
        discussPostRepository.save(discussPost);
    }

    /**
     * 删除数据
     */
    @Test
    public void testDelete() {
//        discussPostRepository.delete(discussPostMapper.selectDiscussPostById(19));
        // 删除所有
        discussPostRepository.deleteAll();
    }

    /**
     * 搜索
     */
    @Test
    public void search() {
        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                .withQuery(QueryBuilders.multiMatchQuery("我", "title", "content"))
                .withSort(SortBuilders.fieldSort("type").order(SortOrder.DESC))
                .withSort(SortBuilders.fieldSort("score").order(SortOrder.DESC))
                .withSort(SortBuilders.fieldSort("createTime").order(SortOrder.DESC))
                .withPageable(PageRequest.of(0, 10))
                .withHighlightFields(
                        new HighlightBuilder.Field("title").preTags("<em>").postTags("</em>"),
                        new HighlightBuilder.Field("content").preTags("<em>").postTags("</em>")
                ).build();

        // 不完善，没有高亮显示
        Page<DiscussPost> page = discussPostRepository.search(searchQuery);

    }

    @Test
    public void testSearchByTemplate() {
        SearchQuery searchQuery = new NativeSearchQueryBuilder()
                .withQuery(QueryBuilders.multiMatchQuery("我", "title", "content"))
                .withSort(SortBuilders.fieldSort("type").order(SortOrder.DESC))
                .withSort(SortBuilders.fieldSort("score").order(SortOrder.DESC))
                .withSort(SortBuilders.fieldSort("createTime").order(SortOrder.DESC))
                .withPageable(PageRequest.of(0, 10))
                .withHighlightFields(
                        new HighlightBuilder.Field("title").preTags("<em>").postTags("</em>"),
                        new HighlightBuilder.Field("content").preTags("<em>").postTags("</em>")
                ).build();

        Page<DiscussPost> page = elasticsearchTemplate.queryForPage(searchQuery, DiscussPost.class, new SearchResultMapper() {
            @Override
            public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> aClass, Pageable pageable) {
                SearchHits hits = response.getHits();
                if (hits.getTotalHits() <= 0) {
                    return null;
                }
                List<DiscussPost> list = new ArrayList<>();
                for (SearchHit hit : hits) {
                    DiscussPost post = new DiscussPost();
                    // 获取id
                    String id = hit.getSourceAsMap().get("id").toString();
                    post.setId(Integer.valueOf(id));

                    // 获取userid
                    String userId = hit.getSourceAsMap().get("userId").toString();
                    post.setUserId(Integer.valueOf(userId));

                    // 获取title
                    String title = hit.getSourceAsMap().get("title").toString();
                    post.setTitle(title);

                    // 获取content
                    String content = hit.getSourceAsMap().get("content").toString();
                    post.setContent(content);

                    // 获取status
                    String status = hit.getSourceAsMap().get("status").toString();
                    post.setStatus(Integer.valueOf(status));

                    // 获取createTime
                    String createTime = hit.getSourceAsMap().get("createTime").toString();
                    post.setCreateTime(new Date(Long.valueOf(createTime)));

                    // 获取commentCount
                    String commentCount = hit.getSourceAsMap().get("commentCount").toString();
                    post.setCommentCount(Integer.valueOf(commentCount));

                    // 处理高亮显示的结果
                    HighlightField titleFields = hit.getHighlightFields().get("title");
                    if (titleFields != null) {
                        post.setTitle(titleFields.getFragments()[0].toString());
                    }
                    HighlightField contentFields = hit.getHighlightFields().get("content");
                    if (contentFields != null) {
                        post.setContent(contentFields.getFragments()[0].toString());
                    }

                    list.add(post);
                }
                return new AggregatedPageImpl(list, pageable, hits.getTotalHits(),
                        response.getAggregations(), response.getScrollId(), hits.getMaxScore());
            }
        });
        System.out.println(page.getTotalElements());
        System.out.println(page.getTotalPages());
        System.out.println(page.getNumber());
        System.out.println(page.getSize());
        for (DiscussPost discussPost : page) {
            System.out.println(discussPost);
        }
    }

}

3、开发社区搜索功能

3.1 搜索服务

ElasticsearchService

1. 将帖子保存至Elasticsearch服务器

// 添加一个对象
public void saveDiscussPost(DiscussPost post){
    discussPostRepository.save(post);
}

2. 从Elasticsearch服务器删除帖子

// 根据id删除
public void deleteDiscussPost(int id){
    discussPostRepository.deleteById(id);
}

3. 从Elasticsearch服务器搜索帖子

// 搜索并高亮显示
public Page<DiscussPost> searchDiscussPost(String keyword,int current,int limit){
    SearchQuery searchQuery = new NativeSearchQueryBuilder()
            .withQuery(QueryBuilders.multiMatchQuery(keyword, "title", "content"))
            .withSort(SortBuilders.fieldSort("type").order(SortOrder.DESC))
            .withSort(SortBuilders.fieldSort("score").order(SortOrder.DESC))
            .withSort(SortBuilders.fieldSort("createTime").order(SortOrder.DESC))
            .withPageable(PageRequest.of(current, limit))
            .withHighlightFields(
                    new HighlightBuilder.Field("title").preTags("<em>").postTags("</em>"),
                    new HighlightBuilder.Field("content").preTags("<em>").postTags("</em>")
            ).build();

    return elasticsearchTemplate.queryForPage(searchQuery, DiscussPost.class, new SearchResultMapper() {
        @Override
        public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> aClass, Pageable pageable) {
            SearchHits hits = response.getHits();
            if (hits.getTotalHits() <= 0) {
                return null;
            }
            List<DiscussPost> list = new ArrayList<>();
            for (SearchHit hit : hits) {
                DiscussPost post = new DiscussPost();
                // 获取id
                String id = hit.getSourceAsMap().get("id").toString();
                post.setId(Integer.valueOf(id));

                // 获取userid
                String userId = hit.getSourceAsMap().get("userId").toString();
                post.setUserId(Integer.valueOf(userId));

                // 获取title
                String title = hit.getSourceAsMap().get("title").toString();
                post.setTitle(title);

                // 获取content
                String content = hit.getSourceAsMap().get("content").toString();
                post.setContent(content);

                // 获取status
                String status = hit.getSourceAsMap().get("status").toString();
                post.setStatus(Integer.valueOf(status));

                // 获取createTime
                String createTime = hit.getSourceAsMap().get("createTime").toString();
                post.setCreateTime(new Date(Long.parseLong(createTime)));

                // 获取commentCount
                String commentCount = hit.getSourceAsMap().get("commentCount").toString();
                post.setCommentCount(Integer.valueOf(commentCount));

                // 处理高亮显示的结果
                HighlightField titleFields = hit.getHighlightFields().get("title");
                if (titleFields != null) {
                    post.setTitle(titleFields.getFragments()[0].toString());
                }
                HighlightField contentFields = hit.getHighlightFields().get("content");
                if (contentFields != null) {
                    post.setContent(contentFields.getFragments()[0].toString());
                }

                list.add(post);
            }
            return new AggregatedPageImpl(list, pageable, hits.getTotalHits(),
                    response.getAggregations(), response.getScrollId(), hits.getMaxScore());
        }
    });
}

3.2 发布事件

1. 发布帖子时，将帖子一部的提交到Elasticsearch服务器

// 触发发帖事件
Event event = new Event()
        .setTopic(TOPIC_PUBLISH)
        .setUserId(user.getId())
        .setEntityType(ENTITY_TYPE_POST)
        .setEntityId(post.getId());
eventProducer.fireEvent(event);

2. 增加评论时，将帖子异步的提交到Elasticsearch服务器

// 触发发布事件
if (comment.getEntityType()==ENTITY_TYPE_POST){
    event = new Event()
            .setTopic(TOPIC_PUBLISH)
            .setUserId(comment.getUserId())
            .setEntityType(ENTITY_TYPE_POST)
            .setEntityId(discussPostId);

    eventProducer.fireEvent(event);
}

3. 在消费组件中增加了一个方法，消费帖子发布事件

// 消费发帖事件
@KafkaListener(topics = {TOPIC_PUBLISH})
public void handlePublishMessage(ConsumerRecord record) {
    if (record == null || record.value() == null) {
        logger.error("消息的内容为空!");
        return;
    }

    Event event = JSONObject.parseObject(record.value().toString(), Event.class);
    if (event == null) {
        logger.error("消息格式错误！");
        return;
    }
    DiscussPost post = discussPostService.findDiscussPostById(event.getEntityId());
    elasticsearchService.saveDiscussPost(post);
}

3.3 显示结果

在控制器中处理搜索请求，在HTML上显示搜索结果

SearchController

@RequestMapping(value = "/search", method = RequestMethod.GET)
    public String search(String keyword, Page page, Model model) {

        // 搜索帖子
        org.springframework.data.domain.Page<DiscussPost> searchResult =
                elasticsearchService.searchDiscussPost(keyword, page.getCurrent() - 1, page.getLimit());

        // 聚合数据
        List<Map<String, Object>> discussPosts = new ArrayList<>();
        if (searchResult != null) {
            for (DiscussPost post : searchResult) {
                Map<String, Object> map = new HashMap<>();
                // 帖子
                map.put("post", post);
                // 作者
                map.put("user", userService.findUserById(post.getUserId()));
                // 点赞数量
                map.put("likeCount", likeService.findEntityLikeCount(ENTITY_TYPE_POST, post.getId()));

                discussPosts.add(map);
            }
        }
        model.addAttribute("discussPosts", discussPosts);
        model.addAttribute("keyword", keyword);

        // 分页信息
        page.setPath("/search?keyword" + keyword);
        page.setRows(searchResult == null ? 0 : (int) searchResult.getTotalElements());

        return "/site/search";
    }

index.html 搜索功能

search.html 页面展示