SpringBoot整合Canal实现MySQL与ES数据同步

开始之前请确认docker中已运行mysql与canal容器，并完成了监听binlog配置
未完成可移步： Docker部署Canal监听MySQL的binlog

SpringBoot项目

本次在SpringBoot整合Easy-ES实现对ES的基础操作项目基础上进行操作
此部分操作请移步：SpringBoot整合Easy-ES实现对ES操作

引入Canal依赖

        <dependency><groupId>top.javatool</groupId><artifactId>canal-spring-boot-starter</artifactId><version>1.2.1-RELEASE</version></dependency>

配置文件

新增以下内容
注意修改server，换成自己的canal地址，端口号

canal:server: canal地址:11111destination: example

项目结构

设置监听类

CanalTable注解是监听的表名，实现EntryHandler接口
重写监听到mysql增删改操作时，这里的进行自定义操作，方法也都是通过Easy-ES实现

@CanalTable("document")
@Component
public class DocumentHandler implements EntryHandler<Document> {@Resourceprivate IDocumentService documentService;/*** mysql中数据有新增时自动执行* @param document 新增的数据*/@Overridepublic void insert(Document document) {try {documentService.addData(document);} catch (Exception e) {e.printStackTrace();}}/*** mysql中数据有修改时自动执行* @param before 修改前的数据* @param after 修改后的数据*/@Overridepublic void update(Document before, Document after) {documentService.updateData(after);}/*** mysql中数据有删除时自动执行* @param document 要删除的数据*/@Overridepublic void delete(Document document) {documentService.deleteData(document);}
}

其余类、接口内容

启动类

添加扫描ESMapper的注解，指定路径

@EsMapperScan("com.mine.easyEs.mapper")

实体类

@Data
public class Document {@Id/*** es中的唯一id*/private String id;/*** 文档标题*/private String title;/*** 文档内容*/private String content;/*** 创建时间*/private Date createTime;
}

Controller类

包括对索引操作和对数据进行操作的接口

@RestController
@RequestMapping("/ee")
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class DocumentController {private final IDocumentService documentService;/*** 创建索引* @return 结果信息* @throws Exception*/@GetMapping("/createIndex")public String createIndex() throws Exception {return documentService.createIndex();}/*** 删除索引* @return 结果信息*/@GetMapping("/deleteIndex")public String deleteIndex(){return documentService.deleteIndex();}/*** 查询ES所有数据* @return 查询Document结果对象集合*/@GetMapping("/findAll")public List<Document> findAll(){return documentService.findAllData();}/*** ES新增数据* @param document 新增数据对象* @return 结果信息* @throws Exception*/@GetMapping("/add")public String addData(Document document) throws Exception {return documentService.addData(document);}/*** 修改ES数据* @param document 修改数据对象*/@GetMapping("/update")public String updateData(Document document){return documentService.updateData(document);}/*** 根据id删除ES数据* @param id 需要删除的数据的id* @return*/@GetMapping("/delete")public String deleteData(String id){return documentService.deleteDataById(id);}/*** 分词匹配查询content字段* @param value 查询内容* @return*/@GetMapping("/match")public List<Document> findMatch(String value){return documentService.findMatch(value);}}

Mapper接口

继承BaseMapper,整体操作都与MybatisPlus类似

public interface DocumentMapper extends BaseEsMapper<Document> {
}

Serice接口

public interface IDocumentService {/*** 查询ES所有数据* @return 查询Document结果对象集合*/List<Document> findAllData();/*** 创建索引* @return 结果信息* @throws Exception*/String createIndex() throws Exception;/*** 删除索引* @return 结果信息*/String deleteIndex();/*** ES新增数据* @param document 新增数据实体类* @return 结果信息* @throws Exception*/String addData(Document document) throws Exception;/*** 根据id删除ES数据* @param id 需要删除的数据的id* @return*/String deleteDataById(String id);String deleteData(Document document);/*** 修改ES数据* @param document 修改数据对象*/String updateData(Document document);/*** 分词匹配查询content字段* @param value 查询内容* @return*/List<Document> findMatch(String value);
}

Service实现类

@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class DocumentServiceImpl implements IDocumentService {private final DocumentMapper documentMapper;/*** 查询ES所有数据* @return 查询Document结果对象集合*/@Overridepublic List<Document> findAllData() {LambdaEsQueryWrapper<Document> wrapper = new LambdaEsQueryWrapper<>();wrapper.matchAllQuery();return documentMapper.selectList(wrapper);}/*** 创建索引* @return 结果信息* @throws Exception*/@Overridepublic String createIndex() throws Exception {StringBuilder msg = new StringBuilder();String indexName = Document.class.getSimpleName().toLowerCase();boolean existsIndex = documentMapper.existsIndex(indexName);if (existsIndex){throw new Exception("Document实体对应索引已存在,删除索引接口：deleteIndex");}boolean success = documentMapper.createIndex();if (success){msg.append("Document索引创建成功");}else {msg.append("索引创建失败");}return msg.toString();}/*** 删除索引* @return 结果信息*/@Overridepublic String deleteIndex() {StringBuilder msg = new StringBuilder();String indexName = Document.class.getSimpleName().toLowerCase();if (documentMapper.deleteIndex(indexName)){msg.append("删除成功");}else {msg.append("删除失败");}return msg.toString();}/*** ES新增数据* @param document 新增数据实体类* @return 结果信息* @throws Exception*/@Overridepublic String addData(Document document) throws Exception {if (StringUtils.isEmpty(document.getTitle()) || StringUtils.isEmpty(document.getContent())) {throw new Exception("请补全title及content数据");}document.setCreateTime(new Date());documentMapper.insert(document);return "Added successfully！";}/*** 根据id删除ES数据* @param id 需要删除的数据的id* @return*/@Overridepublic String deleteDataById(String id) {documentMapper.deleteById(id);return "Success";}@Overridepublic String deleteData(Document document) {documentMapper.deleteById(document.getId());return "Success";}/*** 修改ES数据* @param document 修改数据对象*/@Overridepublic String updateData(Document document) {documentMapper.updateById(document);return "Success";}/*** 分词匹配查询content字段* @param value 查询内容* @return*/@Overridepublic List<Document> findMatch(String value) {LambdaEsQueryWrapper<Document> wrapper = new LambdaEsQueryWrapper<>();wrapper.match(Document::getContent,value);wrapper.orderByDesc(Document::getCreateTime);List<Document> documents = documentMapper.selectList(wrapper);return documents;}
}

运行

可以看到，正在监听，只不过目前我们没有对数据库进行操作。

测试

我们在数据库新增一条数据

此时插入的这条数据被监听到了

通过测试方法查看ES中是否插入了这条数据

@Testpublic void testSelect() {// 测试查询String title = "3";Document document = EsWrappers.lambdaChainQuery(documentMapper).eq(Document::getTitle, title).one();System.out.println(document);Assertions.assertEquals(title,document.getTitle());}

查到了在mysql新插入的这条数据
数据同步成功