SpringBoot+Mybatis-Plus+Elasticsearch + RabbitMQ實現關鍵字搜索高亮展示及數據同步

一、概述&介紹

Elasticsearch：

Elasticsearch 是基於Lucense 技術的搜索引擎（服務器），將數據進行緩存再進行查詢。

​ 與數據庫查詢的比較：

​ （1）相當於sql查詢的 like 模糊查詢，但Elasticsearch支持分詞模糊查詢，比如字符串 “abcdef你 好abdcd” ，通過數據庫查詢 [select * from user where user_name like '%你 好%'; ]只能查詢僅限於以“你 好”為整體得到相關的結果【abcdef你 好abdcd】或【abcdef你 好】或【你 好abdcd】等。而Elasticsearch搜索結果將“你 好”進行拆分查詢，結果可以得到【abcdef你 好abdcd】【abcdef你】、【好abdcd】、【 好abd】，【ef你】等，可見查詢效果更靈活范圍更廣。

RabbitMQ：

MQ全稱為Message Queue, 消息隊列（MQ）是一種應用程序對應用程序的通信方法。應用程序通過讀寫出入隊列的消息（針對應用程序的數據）來通信，而無需專用連接來鏈接它們。消息傳遞指的是程序之間通過在消息中發送數據進行通信，而不是通過直接調用彼此來通信，直接調用通常是用於諸如遠程過程調用的技術。排隊指的是應用程序通過 隊列來通信。隊列的使用除去了接收和發送應用程序同時執行的要求。

RabbitMQ是使用Erlang語言開發的開源消息隊列系統，基於AMQP協議來實現。AMQP的主要特征是面向消息、隊列、路由(包括點對點和發布/訂閱)、可靠性、 安全。AMQP協議更多用在企業系統內，對數據一致性、穩定性和可靠性要求很高的場景，對性能和吞吐量的要求還在其次。

二、使用場景：

Elasticsearch 使用場景：網站全局搜索、電商網站商品推薦、文章內容檢索、文本分析等等。

RabbitMQ 使用場景：

1. 解耦（為面向服務的架構（SOA）提供基本的最終一致性實現）
2. 異步提升效率
3. 流量削峰

官網：https://www.elastic.co/cn/

下載地址：https://www.elastic.co/cn/downloads/elasticsearch

三、環境描述：

技術架構：

后端：Springboot、Mybtis-Plus、Elasticsearch、RabbitMQ

前端：Freemark

四、環境搭建：

具體安裝方式可以參考以下，本文不做過多講解

Elasticsearch安裝：

windows版本安裝：https://blog.csdn.net/chen_2890/article/details/83757022

linux版本安裝：https://blog.csdn.net/qq_32502511/article/details/86140486

啟動系統變量限制問題參考https://www.cnblogs.com/zuikeol/p/10930685.html

RabbitMQ安裝：

windows版本安裝：https://blog.csdn.net/zhm3023/article/details/82217222

linux版本安裝：https://www.cnblogs.com/rmxd/p/11583932.html

五、具體實現

本文實現為：

1. 網站文章搜索，搜索內容根據標題、內容、文章描述進行搜索，實現分頁搜索
2. 發布文章數據異步同步到ES。

實現步驟描述：

1. 與SpringBoot整合；

• pom.xml導入maven依賴包

<!-- springdata整合elasticsearch -->
<dependency>
	<groupId>org.springframework.data</groupId>
	<artifactId>spring-data-elasticsearch</artifactId>
</dependency>
<!--整合rabbitmq-->
<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

• application.yml配置

spring:
	#elasticsearch 配置
	data:
		elasticsearch:
			cluster-name: elasticsearch
			cluster-nodes: 127.0.0.1:9300
			repositories:
				enabled: true
	#rabbitmq 配置
    rabbitmq:
      username: mblog
      password: mblog
      host: 127.0.0.1
      port: 5672			

2. 新增文章時，同步數據到elasticsearch搜索引擎服務器中；

文章數據表結構：

DROP TABLE IF EXISTS `mto_post`;
CREATE TABLE `mto_post` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `author_id` bigint(20) DEFAULT NULL,
  `channel_id` int(11) DEFAULT NULL,
  `comments` int(11) NOT NULL,
  `created` datetime DEFAULT NULL,
  `favors` int(11) NOT NULL,
  `featured` int(11) NOT NULL,
  `status` int(11) NOT NULL,
  `summary` varchar(140) DEFAULT NULL,
  `tags` varchar(64) DEFAULT NULL,
  `thumbnail` varchar(128) DEFAULT NULL,
  `title` varchar(64) DEFAULT NULL,
  `views` int(11) NOT NULL,
  `weight` int(11) NOT NULL,
  PRIMARY KEY (`id`),
  KEY `IK_CHANNEL_ID` (`channel_id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

數據同步到Elasticsearch搜索引擎服務器：

@Service
public class PostServiceImpl implements PostService {

	@Autowired
    private PostMapper postMapper;
	@Autowired
    private PostAttributeMapper postAttributeMapper;
	@Autowired
    private TagService tagService;
	@Autowired
    private RabbitTemplate rabbitTemplate;
	
	@Override
	@Transactional
	public long post(PostVO post) {
		Post po = new Post();
		BeanUtils.copyProperties(post, po);
		po.setStatus(post.getStatus());
		// 處理摘要
		if (StringUtils.isBlank(post.getSummary())) {
			po.setSummary(trimSummary(post.getEditor(), post.getContent()));
		} else {
			po.setSummary(post.getSummary());
		}
		postMapper.insert(po);

		tagService.batchUpdate(po.getTags(), po.getId());

		String key = ResourceLock.getPostKey(po.getId());
		AtomicInteger lock = ResourceLock.getAtomicInteger(key);
		try {
			synchronized (lock){
				PostAttribute attr = new PostAttribute();
				attr.setContent(post.getContent());
				attr.setEditor(post.getEditor());
				attr.setPostId(po.getId());
				postAttributeMapper.insert(attr);

				countResource(po.getId(), null,  attr.getContent());
				onPushEvent(po, PostUpdateEvent.ACTION_PUBLISH);

				//使用rabbitmq同步到elasticsearch搜索引擎服務器
				rabbitmqSend(po, ESMqMessage.CREATE_OR_UPDATE);
				return po.getId();
			}
		}finally {
			ResourceLock.giveUpAtomicInteger(key);
		}
	}
	
	/**
     * rabbitmq發送
     *
     * @param po   文章實體對象
     * @param type 類型：CREATE_OR_UPDATE 創建or更新索引；REMOVE 刪除索引
     */
    private void rabbitmqSend(Post po, String type) {
        rabbitTemplate.convertAndSend(RabbitConstant.ES_EXCHAGE, RabbitConstant.ES_ROUTING_KEY,
                        new ESMqMessage(po.getId(), type));
    }
}

/**
 * @ClassName: RabbitConstant
 * @Auther: Jerry
 * @Date: 2020/5/15 9:23
 * @Desctiption: rabbit常量
 * @Version: 1.0
 */
public class RabbitConstant {
	/**es同步隊列*/
    public final static String ES_QUEUE = "es_queue";
    public final static String ES_EXCHAGE = "es_exchage";
    public final static String ES_ROUTING_KEY = "es_routing_key";
}

/**
 * @ClassName: ESMqMessage
 * @Auther: Jerry
 * @Date: 2020/5/14 16:58
 * @Desctiption: 文章相關消息隊列
 * @Version: 1.0
 */
@Data
@AllArgsConstructor
public class ESMqMessage implements Serializable {
    private static final long serialVersionUID = 3572599349158869479L;
    /**
     * 新增或修改
     */
    public final static String CREATE_OR_UPDATE = "create_or_update";
    /**
     * 刪除
     */
    public final static String REMOVE = "remove";
    /**
     * 文章id
     */
    private long postId;
    /**
     * 文章操作類型
     */
    private String action;
}

@Slf4j
@Component
@RabbitListener(queues = RabbitConstant.ES_QUEUE)
public class ESMqHandler {

    @Autowired
    private PostSearchService postSearchService;

    @RabbitHandler
    public void handler(ESMqMessage message) {
        log.info("PostMqHandler -------> mq 收到一條消息： {}", message.toString());
        switch (message.getAction()) {
            case ESMqMessage.CREATE_OR_UPDATE:
                postSearchService.createOrUpdateIndex(message);
                break;
            case ESMqMessage.REMOVE:
                postSearchService.removeIndex(message);
                break;
            default:
                log.error("沒找到對應的消息類型，請注意！！ --》 {}", message.toString());
                break;
        }
    }
}

實體類：

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
@Document(indexName = "es_article_index", type = "doc",
        useServerConfiguration = true, createIndex = false)
public class Articles implements Serializable {

    private static final long serialVersionUID = -728655685413761417L;

    /**
     * ID
     */
    @Id
    private Long id;
    /**
     * 狀態
     */
    private int status;
    /**
     * 標題
     */
    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    private String title;

    /**
     * 內容
     */
    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    private String summary;
    /**
     * 標簽
     */
    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    private String tags;
    /**
     * 創建時間
     */
    private Date created;
    /**
     * 更新時間
     */
    private Date updated;
    /**
     * 作者id
     */
    private Long authorId;

    /**
     * 作者
     */
    private Object author;

    /**
     * 分組/模塊
     */
    private int channelId;

    /**
     * 分組/模塊
     */
    private Object channel;

    /**
     * 收藏數
     */
    private int favors;
    /**
     * 評論數
     */
    private int comments;
    /**
     * 閱讀數
     */
    private int views;
    /**
     * 推薦狀態
     */
    private int featured;
    /**
     * 預覽圖
     */
    private String thumbnail;
}

搜索接口：

/**
 * @ClassName: ArticlesRepository
 * @Auther: Jerry
 * @Date: 2020/4/20 11:32
 * @Desctiption: 文章搜索
 * @Version: 1.0
 */
public interface ArticlesRepository extends ElasticsearchRepository<Articles, Long> {
}

3. 分頁關鍵詞搜索高亮展示具體實現；

   （1）controller實現：

/**
 * 文章搜索
 * @author langhsu
 *
 */
@Controller
public class SearchController extends BaseController {
	@Autowired
	private PostSearchService postSearchService;

	@RequestMapping("/search")
	public String search(HttpServletRequest request, String kw, ModelMap model) {
		try {
			if (StringUtils.isNotEmpty(kw)) {
				int pageNo = ServletRequestUtils.getIntParameter(request, "pageNo", 1);
				int pageSize = ServletRequestUtils.getIntParameter(request, "pageSize", 10);
				IPage<Articles> page = postSearchService.search(pageNo, pageSize,kw);
				model.put("results", page);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		model.put("kw", kw);
		return view(Views.SEARCH);
	}
	
}

​ （2）service實現：

@Slf4j
@Service
@Transactional(readOnly = true)
public class PostSearchServiceImpl implements PostSearchService {

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;
    @Autowired
    private PostService postService;
	@Autowired
    private ChannelService channelService;
    @Autowired
    private ArticlesRepository articlesRepository;

	@Override
    public IPage<Articles> search(int page, int size, String term) throws Exception {
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery()
                .should(QueryBuilders.matchQuery("title", term))
                .should(QueryBuilders.matchQuery("summary", term))
                .should(QueryBuilders.matchQuery("tags", term));
        // 創建高亮查詢
        NativeSearchQueryBuilder nativeSearchQuery = new NativeSearchQueryBuilder();
        nativeSearchQuery.withQuery(boolQueryBuilder);
        nativeSearchQuery.withHighlightFields(new HighlightBuilder.Field("title"),
                new HighlightBuilder.Field("summary"),
                new HighlightBuilder.Field("tags"));
        nativeSearchQuery.withHighlightBuilder(new HighlightBuilder().preTags("<span style='color:red'>").postTags("</span>"));
        // 設置分頁,頁碼要減1
        nativeSearchQuery.withPageable(PageRequest.of(page - 1, size));
        // 分頁對象
        AggregatedPage<Articles> eSearchPage = elasticsearchTemplate.queryForPage(nativeSearchQuery.build(), Articles.class,
                new SearchResultMapper() {
                    @Override
                    public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {

                        ArrayList<Articles> list = new ArrayList<Articles>();
                        SearchHits hits = response.getHits();
                        for (SearchHit searchHit : hits) {
                            if (hits.getHits().length <= 0) {
                                return null;
                            }
                            Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
                            Integer id = (Integer) sourceAsMap.get("id");
                            String title = (String) sourceAsMap.get("title");
                            Object author = sourceAsMap.get("author");

                            String summary = (String) sourceAsMap.get("summary");
                            String tags = (String) sourceAsMap.get("tags");

                            Object channel = sourceAsMap.get("channel");
                            String thumbnail = (String) sourceAsMap.get("thumbnail");
                            Integer favors = (Integer) sourceAsMap.get("favors");
                            Integer comments = (Integer) sourceAsMap.get("comments");
                            Integer views = (Integer) sourceAsMap.get("views");
                            Integer featured = (Integer) sourceAsMap.get("featured");
                            Date created = new Date((Long) sourceAsMap.get("created"));

                            Articles seArticleVo = new Articles();
                            HighlightField highLightField = searchHit.getHighlightFields().get("title");
                            if (highLightField == null) {
                                seArticleVo.setTitle(title);
                            } else {
                                seArticleVo.setTitle(highLightField.fragments()[0].toString());
                            }
                            highLightField = searchHit.getHighlightFields().get("summary");
                            if (highLightField == null) {
                                seArticleVo.setSummary(summary);
                            } else {
                                seArticleVo.setSummary(highLightField.fragments()[0].toString());
                            }
                            highLightField = searchHit.getHighlightFields().get("tags");
                            if (highLightField == null) {
                                seArticleVo.setTags(tags);
                            } else {
                                seArticleVo.setTags(highLightField.fragments()[0].toString());
                            }
                            highLightField = searchHit.getHighlightFields().get("id");
                            if (highLightField == null) {
                                seArticleVo.setId(id.longValue());
                            } else {
                                seArticleVo.setId(Long.parseLong(highLightField.fragments()[0].toString()));
                            }
                            seArticleVo.setAuthor(author);
                            seArticleVo.setChannel(channel);
                            seArticleVo.setCreated(created);
                            seArticleVo.setThumbnail(thumbnail);
                            seArticleVo.setFavors(favors);
                            seArticleVo.setComments(comments);
                            seArticleVo.setViews(views);
                            seArticleVo.setFeatured(featured == null ? 0 : featured);
                            list.add(seArticleVo);
                        }
                        AggregatedPage<T> pageResult = new AggregatedPageImpl<T>((List<T>) list, pageable, hits.getTotalHits());
                        return pageResult;
                    }
                });
        long pageNum = Long.valueOf(eSearchPage.getNumber());
        long pageSize = Long.valueOf(eSearchPage.getPageable().getPageSize());
        Page page1 = new Page(pageNum, pageSize);
        page1.setRecords(eSearchPage.getContent());
        page1.setTotal(Long.valueOf(eSearchPage.getTotalElements()));
        return page1;
    }
    
    @Override
    public void createOrUpdateIndex(ESMqMessage message) {
        long postId = message.getPostId();
        Post post = postService.getPostById(postId);
        Articles articles = BeanMapUtil.post2Articles(post);
        UserVO author = userService.get(post.getAuthorId());
        Channel channel = channelService.getById(post.getChannelId());
        articles.setAuthor(author);
        articles.setChannel(channel);
        articlesRepository.save(articles);
        log.info("es 索引更新成功！ ---> {}", articles.toString());
    }
    
    @Override
    public void removeIndex(ESMqMessage message) {
        long postId = message.getPostId();

        articlesRepository.deleteById(postId);
        log.info("es 索引刪除成功！ ---> {}", message.toString());
    }
}    

六、總結

使用Elasiticsearch 時需要注意的幾個問題：

（1）分頁需要重新計算頁碼，執行查詢時需要設置nativeSearchQuery.withPageable(new PageRequest(request.getPageNum() - 1, request.getPageSize())); 查詢到結果后需要計算頁碼；

（2）ES查詢結果后，單獨處理關鍵字，命中關鍵字部分通過withHighlightBuilder().preTags方法設置命中文本標記。

​ nativeSearchQuery.withHighlightBuilder(new HighlightBuilder().preTags("<span style='color:red'>").postTags(""));

finally,大功告成！