Springboot+Mybatis+Elasticsearch+RabbitMQ实现数据同步到ES及关键字搜索高亮展示_springboot elasticsearch类似mybatis查询

一、概述&介绍

**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			

1. 新增文章时，同步数据到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> {
}

1. 分页关键词搜索高亮展示具体实现；

   （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,大功告成！