SpringBoot+MybatisPlus+Elasticsearch+RabbitMQ实现关键字搜索高亮展示及数据同步到ES

一、概述&介绍

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依赖包



org.springframework.data
spring-data-elasticsearch



org.springframework.boot
spring-boot-starter-amqp



• 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 {
}


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 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 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("").postTags(""));
// 设置分页,页码要减1
nativeSearchQuery.withPageable(PageRequest.of(page - 1, size));
// 分页对象
AggregatedPage eSearchPage = elasticsearchTemplate.queryForPage(nativeSearchQuery.build(), Articles.class,
new SearchResultMapper() {
@Override
public AggregatedPage mapResults(SearchResponse response, Class clazz, Pageable pageable) {
ArrayList list = new ArrayList();
SearchHits hits = response.getHits();
for (SearchHit searchHit : hits) {
if (hits.getHits().length <= 0) {
return null;
}
Map 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 pageResult = new AggregatedPageImpl((List) 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("

finally,大功告成！