開始之前請確認docker中已運行mysql與canal容器，并完成了監(jiān)聽binlog配置
未完成可移步： Docker部署Canal監(jiān)聽MySQL的binlog

SpringBoot項目

本次在SpringBoot整合Easy-ES實現(xiàn)對ES的基礎(chǔ)操作項目基礎(chǔ)上進行操作
此部分操作請移步：SpringBoot整合Easy-ES實現(xiàn)對ES操作

引入Canal依賴

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

配置文件

新增以下內(nèi)容
注意修改server，換成自己的canal地址，端口號

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

項目結(jié)構(gòu)

設(shè)置監(jiān)聽類

CanalTable注解是監(jiān)聽的表名，實現(xiàn)EntryHandler接口
重寫監(jiān)聽到mysql增刪改操作時，這里的進行自定義操作，方法也都是通過Easy-ES實現(xiàn)

@CanalTable("document")
@Component
public class DocumentHandler implements EntryHandler<Document> {@Resourceprivate IDocumentService documentService;/*** mysql中數(shù)據(jù)有新增時自動執(zhí)行* @param document 新增的數(shù)據(jù)*/@Overridepublic void insert(Document document) {try {documentService.addData(document);} catch (Exception e) {e.printStackTrace();}}/*** mysql中數(shù)據(jù)有修改時自動執(zhí)行* @param before 修改前的數(shù)據(jù)* @param after 修改后的數(shù)據(jù)*/@Overridepublic void update(Document before, Document after) {documentService.updateData(after);}/*** mysql中數(shù)據(jù)有刪除時自動執(zhí)行* @param document 要刪除的數(shù)據(jù)*/@Overridepublic void delete(Document document) {documentService.deleteData(document);}
}

其余類、接口內(nèi)容

啟動類

添加掃描ESMapper的注解，指定路徑

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

實體類

@Data
public class Document {@Id/*** es中的唯一id*/private String id;/*** 文檔標題*/private String title;/*** 文檔內(nèi)容*/private String content;/*** 創(chuàng)建時間*/private Date createTime;
}

Controller類

包括對索引操作和對數(shù)據(jù)進行操作的接口

@RestController
@RequestMapping("/ee")
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class DocumentController {private final IDocumentService documentService;/*** 創(chuàng)建索引* @return 結(jié)果信息* @throws Exception*/@GetMapping("/createIndex")public String createIndex() throws Exception {return documentService.createIndex();}/*** 刪除索引* @return 結(jié)果信息*/@GetMapping("/deleteIndex")public String deleteIndex(){return documentService.deleteIndex();}/*** 查詢ES所有數(shù)據(jù)* @return 查詢Document結(jié)果對象集合*/@GetMapping("/findAll")public List<Document> findAll(){return documentService.findAllData();}/*** ES新增數(shù)據(jù)* @param document 新增數(shù)據(jù)對象* @return 結(jié)果信息* @throws Exception*/@GetMapping("/add")public String addData(Document document) throws Exception {return documentService.addData(document);}/*** 修改ES數(shù)據(jù)* @param document 修改數(shù)據(jù)對象*/@GetMapping("/update")public String updateData(Document document){return documentService.updateData(document);}/*** 根據(jù)id刪除ES數(shù)據(jù)* @param id 需要刪除的數(shù)據(jù)的id* @return*/@GetMapping("/delete")public String deleteData(String id){return documentService.deleteDataById(id);}/*** 分詞匹配查詢content字段* @param value 查詢內(nèi)容* @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所有數(shù)據(jù)* @return 查詢Document結(jié)果對象集合*/List<Document> findAllData();/*** 創(chuàng)建索引* @return 結(jié)果信息* @throws Exception*/String createIndex() throws Exception;/*** 刪除索引* @return 結(jié)果信息*/String deleteIndex();/*** ES新增數(shù)據(jù)* @param document 新增數(shù)據(jù)實體類* @return 結(jié)果信息* @throws Exception*/String addData(Document document) throws Exception;/*** 根據(jù)id刪除ES數(shù)據(jù)* @param id 需要刪除的數(shù)據(jù)的id* @return*/String deleteDataById(String id);String deleteData(Document document);/*** 修改ES數(shù)據(jù)* @param document 修改數(shù)據(jù)對象*/String updateData(Document document);/*** 分詞匹配查詢content字段* @param value 查詢內(nèi)容* @return*/List<Document> findMatch(String value);
}

Service實現(xiàn)類

@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class DocumentServiceImpl implements IDocumentService {private final DocumentMapper documentMapper;/*** 查詢ES所有數(shù)據(jù)* @return 查詢Document結(jié)果對象集合*/@Overridepublic List<Document> findAllData() {LambdaEsQueryWrapper<Document> wrapper = new LambdaEsQueryWrapper<>();wrapper.matchAllQuery();return documentMapper.selectList(wrapper);}/*** 創(chuàng)建索引* @return 結(jié)果信息* @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實體對應(yīng)索引已存在,刪除索引接口：deleteIndex");}boolean success = documentMapper.createIndex();if (success){msg.append("Document索引創(chuàng)建成功");}else {msg.append("索引創(chuàng)建失敗");}return msg.toString();}/*** 刪除索引* @return 結(jié)果信息*/@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新增數(shù)據(jù)* @param document 新增數(shù)據(jù)實體類* @return 結(jié)果信息* @throws Exception*/@Overridepublic String addData(Document document) throws Exception {if (StringUtils.isEmpty(document.getTitle()) || StringUtils.isEmpty(document.getContent())) {throw new Exception("請補全title及content數(shù)據(jù)");}document.setCreateTime(new Date());documentMapper.insert(document);return "Added successfully！";}/*** 根據(jù)id刪除ES數(shù)據(jù)* @param id 需要刪除的數(shù)據(jù)的id* @return*/@Overridepublic String deleteDataById(String id) {documentMapper.deleteById(id);return "Success";}@Overridepublic String deleteData(Document document) {documentMapper.deleteById(document.getId());return "Success";}/*** 修改ES數(shù)據(jù)* @param document 修改數(shù)據(jù)對象*/@Overridepublic String updateData(Document document) {documentMapper.updateById(document);return "Success";}/*** 分詞匹配查詢content字段* @param value 查詢內(nèi)容* @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;}
}

運行

可以看到，正在監(jiān)聽，只不過目前我們沒有對數(shù)據(jù)庫進行操作。

測試

我們在數(shù)據(jù)庫新增一條數(shù)據(jù)

此時插入的這條數(shù)據(jù)被監(jiān)聽到了

通過測試方法查看ES中是否插入了這條數(shù)據(jù)

@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新插入的這條數(shù)據(jù)
數(shù)據(jù)同步成功