项目1在线交流平台-6.Elasticsearch分布式搜索引擎-3.ES结合Kafka应用-开发社区搜索功能

参考牛客网高级项目教程
狂神说Elasticsearch教程笔记
尚硅谷Elasticsearch教程笔记

功能需求

• 1.在业务层处理好搜索帖子的服务
  • 包括保存帖子到ES服务器
  • 从服务器中删除帖子
  • 从服务器中查询帖子
• 2.发布事件
  • 在controller层，结合kafka，发布帖子、增加评论时，数据放入消息队列
  • 异步消费消息，将数据同步到ES服务器
• 3.处理模板页面的显示，搜索帖子时，根据关键字显示出满足条件的帖子列表

一、Service层处理操作ES服务器的数据

• 向服务器添加一条帖子
• 删除一条帖子
• 根据关键字查询帖子列表
• 注意，keyword本身为字符串，不能再加”“号

@Service
public class ElasticSearchService {
    @Autowired
    DiscussPostMapper discussPostMapper;
    @Autowired
    DiscussPostRepository repository;
    @Autowired
    ElasticsearchTemplate elasticTemplate;

    /**
     * 向服务器添加一条帖子
     */
    public void savePost(DiscussPost discussPost) {
        repository.save(discussPost);
    }

    /**
     * 删除一条帖子
     */
    public void deletePost(DiscussPost discussPost) {
        repository.delete(discussPost);
    }

    /**
     * 根据关键字查询帖子列表
     */
    public Page<DiscussPost> searchPosts(String keyword, int current, int limit) {
        // 1.构建查询条件
        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();
        // 2.处理查询的结果-拼接高亮显示的部分
        return elasticTemplate.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();

                    String id = hit.getSourceAsMap().get("id").toString();
                    post.setId(Integer.valueOf(id));

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

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

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

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

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

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

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

                    HighlightField contentField = hit.getHighlightFields().get("content");
                    if (contentField != null) {
                        post.setContent(contentField.getFragments()[0].toString());
                    }

                    list.add(post);
                }

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

二、Controller层处理帖子添加和评论事件请求

1.添加帖子时-触发事件-发布消息

// 发布帖子后，触发发帖事件-向kafka服务器发布消息
Event event = new Event()
        .setTopic(TOPIC_PUBLISH)
        .setFromUserId(user.getId())
        .setEntityType(ENTITY_TYPE_POST)
        .setEntityId(discussPost.getId());
eventProducer.sendEvent(event);

2. 添加评论时-触发发帖事件-发布消息

• 因为对帖子发布评论后，帖子的字段需要更新，为保证数据一致性，需要重新发布事件

if (comment.getEntityType() == ENTITY_TYPE_POST) {
    // 触发发帖事件
    event = new Event()
            .setTopic(TOPIC_PUBLISH)
            .setFromUserId(comment.getUserId())
            .setEntityType(ENTITY_TYPE_POST)
            .setEntityId(discussPostId);
    eventProducer.sendEvent(event);
}

三、kafka消费者订阅消息并提交到ES服务器

/**
 * 消费发布帖子的消息-将帖子数据提交到ES服务器
 * @param record
 */
@KafkaListener(topics = TOPIC_PUBLISH)
public void handlePublishMessage(ConsumerRecord record) {
    // 1.边界条件：先检查有无取到消息
    if (record == null || record.value() == null) {
        logger.error("消息的内容为空!");
        return;
    }
    // 2.将拿到的消息恢复成Object类型，方便操作
    Event event = JSONObject.parseObject(record.value().toString(), Event.class);
    if(event == null) {
        logger.error("消息的格式错了!");
        return;
    }

    // 3. 将拿到的消息添加到ES服务器中
    DiscussPost discussPost = discussPostService.selectPostById(event.getEntityId());
    elasticSearchService.savePost(discussPost);
}

四、处理搜索结果

1. Controller处理搜索请求

• 搜索帖子-分页搜索-页面默认设置当前页为1，limit为10

• 注意-需要的分页信息是-当前页码和当前页数量，

  • 与mysql查询的分页条件不同（起始行-当前页显示多少行）

  

/**
 * 根据关键字搜索查询帖子请求
 * @param keyword
 * @param model
 * @param page
 * @return
 */
@RequestMapping(value = "/search", method = RequestMethod.GET)
public String search(String keyword, Model model, Page page) {
    // 搜索帖子-分页搜索-页面默认设置当前页为1，limit为10
    // 注意-需要的分页信息是-当前页码和当前页数量，与mysql查询的分页条件不同（起始行-当前页显示多少行）
    org.springframework.data.domain.Page<DiscussPost> searchResult
            = elasticSearchService.searchPosts(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";
}

2. 处理模板页面

主页头部的搜索栏链接

<!-- 搜索 -->
<form class="form-inline my-2 my-lg-0" method="get" th:action="@{/search}">
   <input class="form-control mr-sm-2" type="search" aria-label="Search" name="keyword" th:value="${keyword}"/>
   <button class="btn btn-outline-light my-2 my-sm-0" type="submit">搜索</button>
</form>

搜索网页设置

<li class="media pb-3 pt-3 mb-3 border-bottom" th:each="map:${discussPosts}">
   <img th:src="${map.user.headerUrl}" class="mr-4 rounded-circle" style="width:50px;height:50px;" alt="用户头像">
   <div class="media-body">
      <h6 class="mt-0 mb-3">
         <a th:href="@{|/discuss/detail/${map.post.id}|}" th:utext="${map.post.title}">备战<em>春招</em>，面试刷题跟他复习，一个月全搞定！</a>
      </h6>
      <div class="mb-3" th:utext="${map.post.content}">帖子内容
      </div>
      <div class="text-muted font-size-12">
         <u class="mr-3" th:utext="${map.user.username}">寒江雪</u> 发布于
         <b th:text="${#dates.format(map.post.createTime,'yyyy-MM-dd HH:mm:ss')}">2019-04-15 15:32:18</b>
         <ul class="d-inline float-right">
            <li class="d-inline ml-2">赞 <i th:text="${map.likeCount}">11</i></li>
            <li class="d-inline ml-2">|</li>
            <li class="d-inline ml-2">回复 <i th:text="${map.post.commentCount}">7</i></li>
         </ul>
      </div>
   </div>
</li>

测试结果：