Springcloud Alibaba 使用Canal将MySql数据实时同步到Elasticsearch

本篇文章在Springcloud Alibaba使用Canal将Mysql数据实时同步到Redis保证缓存的一致性-CSDN博客

基础上使用canal将mysql数据实时同步到Elasticsearch。

1. 数据库准备

CREATE DATABASE /*!32312 IF NOT EXISTS*/`shop` /*!40100 DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci */ /*!80016 DEFAULT ENCRYPTION='N' */;USE `shop`;/*Table structure for table `sku` */DROP TABLE IF EXISTS `sku`;CREATE TABLE `sku` (`id` VARCHAR(60) NOT NULL COMMENT '商品id',`name` VARCHAR(200) NOT NULL COMMENT 'SKU名称',`price` INT NOT NULL DEFAULT '1' COMMENT '价格（分）',`num` INT DEFAULT '100' COMMENT '库存数量',`image` VARCHAR(200) DEFAULT NULL COMMENT '商品图片',`images` VARCHAR(2000) DEFAULT NULL COMMENT '商品图片列表',`create_time` DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',`update_time` DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,`spu_id` VARCHAR(60) DEFAULT NULL COMMENT 'SPUID',`category_id` INT DEFAULT NULL COMMENT '类目ID',`category_name` VARCHAR(200) DEFAULT NULL COMMENT '类目名称',`brand_id` INT DEFAULT NULL COMMENT '品牌id',`brand_name` VARCHAR(100) DEFAULT NULL COMMENT '品牌名称',`sku_attribute` VARCHAR(200) DEFAULT NULL COMMENT '规格',`status` INT DEFAULT '1' COMMENT '商品状态 1-正常，2-下架，3-删除',PRIMARY KEY (`id`),KEY `cid` (`category_id`),KEY `status` (`status`),KEY `updated` (`update_time`)
) ENGINE=INNODB DEFAULT CHARSET=utf8mb3 COMMENT='商品表';/*Data for the table `sku` */INSERT  INTO `sku`(`id`,`name`,`price`,`num`,`image`,`images`,`create_time`,`update_time`,`spu_id`,`category_id`,`category_name`,`brand_id`,`brand_name`,`sku_attribute`,`status`) VALUES 
('1318594982227025922','华为Mate40 Pro 32G',114,1228,'https://sklll.oss-cn-beijing.aliyuncs.com/secby/af1faf56-b10a-4700-9896-3143a2d1c40f.jpg','https://sklll.oss-cn-beijing.aliyuncs.com/secby/a65bfbe4-21b7-42b2-b5cf-47a9730e0a16.jpg,https://sklll.oss-cn-beijing.aliyuncs.com/secby/fa52ef66-7724-4d6e-bece-15eba0f8f903.jpg,https://sklll.oss-cn-beijing.aliyuncs.com/secby/734f0f17-ac73-45d3-a6bf-83e1569ce887.jpg','2020-10-20 16:48:37','2023-12-29 19:02:16','1318594982147334146',11159,'软件研发',11,'华为','{\"就业薪资\":\"10K起\",\"学习费用\":\"2万\"}',1),
('1318596430360813570','华为Mate40 Pro 32G 1800万像素',112,1227,'https://sklll.oss-cn-beijing.aliyuncs.com/secby/9247d041-e940-426c-8e50-06084b631063.jpg','https://sklll.oss-cn-beijing.aliyuncs.com/secby/5f5b7435-6cf2-4797-8f65-d4abff181390.jpg','2020-10-20 16:54:22','2023-12-29 19:07:47','1318596430293704706',11159,'软件研发',11,'华为','{\"就业薪资\":\"10K起\",\"学习费用\":\"2万\"}',1),
('1318596430398562305','华为Mate40 Pro 128G',111,1226,'https://sklll.oss-cn-beijing.aliyuncs.com/secby/900a3618-9884-4778-bad9-c6c31eaf3eab.jpg','https://sklll.oss-cn-beijing.aliyuncs.com/secby/5f5b7435-6cf2-4797-8f65-d4abff181390.jpg','2020-10-20 16:54:22','2023-12-29 19:11:28','1318596430293704706',11159,'软件研发',11,'华为','{\"就业薪资\":\"10K起\",\"学习费用\":\"2万\"}',1),
('1318599511605563394','格力手机 5G手机',100,1225,'https://sklll.oss-cn-beijing.aliyuncs.com/secby/2b233c6a-5acc-449e-ba3a-70a506100948.jpg','https://sklll.oss-cn-beijing.aliyuncs.com/secby/ffc66a17-edfc-43bb-8f66-431b1e9bf606.jpg','2020-10-20 17:06:37','2023-12-29 19:11:25','1318599511492317185',11159,'软件研发',11,'华为','{\"就业薪资\":\"10K起\",\"学习费用\":\"2万\"}',1),
('1318599511647506433','格力手机 5G手机 红色',789,1224,'https://sklll.oss-cn-beijing.aliyuncs.com/secby/1c1fbfea-af9f-49e7-b89b-35e751874399.jpg','https://sklll.oss-cn-beijing.aliyuncs.com/secby/ffc66a17-edfc-43bb-8f66-431b1e9bf606.jpg','2020-10-20 17:06:37','2020-10-20 17:06:37','1318599511492317185',11159,'软件研发',11,'华为','{\"就业薪资\":\"10K起\",\"学习费用\":\"2万\"}',1),

2. 公共部分

公共包

       <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--MyBatis Plus--><dependency><groupId>com.baomidou</groupId><artifactId>mybatis-plus-boot-starter</artifactId><version>3.3.2</version></dependency><!--MySQL--><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><scope>runtime</scope></dependency><!--Nacos--><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId></dependency><dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId></dependency><dependency><groupId>javax.persistence</groupId><artifactId>persistence-api</artifactId><version>1.0</version><scope>compile</scope></dependency>

实体类Sku

@Column注解

用来标识实体类中属性与数据表中字段的对应关系

name
定义了被标注字段在数据库表中所对应字段的名称；由于驼峰命名法，如果不使用@Column字段，canal在监控数据变化时，获得的实体类部分字段为null，比如create_time等等。

@TableName(value ="sku")
@Data
@Table
public class Sku implements Serializable {@TableId(type = IdType.ASSIGN_ID)private String id;private String name;private Integer price;private Integer num;private String image;private String images;@Column(name = "create_time")private Date createTime;@Column(name = "update_time")private Date updateTime;@Column(name = "spu_id")private String spuId;@Column(name = "category_id")private Integer categoryId;@Column(name = "category_name")private String categoryName;@Column(name = "brand_id")private Integer brandId;@Column(name = "brand_name")private String brandName;@Column(name = "sku_attribute")private String skuAttribute;private Integer status;private static final long serialVersionUID = 1L;
}

实体类SkuEs

@Data
@Document(indexName = "skusearch")//indexName一定全小写，不然出错
public class SkuEs {@Idprivate String id;@Field(type = FieldType.Text,analyzer = "ik_smart",searchAnalyzer = "ik_smart")private String name;private Integer price;private Integer num;private String image;private String images;private Date createTime;private Date updateTime;private String spuId;private Integer categoryId;//Keyword：不分词@Field(type= FieldType.Keyword)private String categoryName;private Integer brandId;@Field(type=FieldType.Keyword)private String brandName;@Field(type=FieldType.Keyword)private String skuAttribute;private Integer status;
}

3. mall-search-service

导入依赖：

       <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId><version>2.6.3</version></dependency>

这里使用的springboot版本为2.7.12，springboot和spring-boot-starter-data-elasticsearch的版本问题会导致异常：

java.lang.NoSuchFieldError: INDEX_CONTENT_TYPE

bootstrap.yaml代码如下：

server:port: 8084
spring:application:name: mall-searchcloud:nacos:config:file-extension: yamlserver-addr: localhost:8848discovery:#Nacos的注册地址server-addr: localhost:8848#Elasticsearch服务配置 6.8.12elasticsearch:uris: http://localhost:9200

编写SkuSearchMapper,在主启动类上添加@EnableElasticsearchRepositories(basePackages = {"xx"})，basePackages指定mapper包路径

public interface SkuSearchMapper extends ElasticsearchRepository<SkuEs,String> {
}

Service层

public interface SkuSearchService {void add(SkuEs skuEs);void del(String id);
}@Service
public class SkuSearchServiceImpl implements SkuSearchService {@AutowiredSkuSearchMapper skuSearchMapper;@AutowiredElasticsearchRestTemplate elasticsearchRestTemplate;@Overridepublic void add(SkuEs skuEs) {skuSearchMapper.save(skuEs);}@Overridepublic void del(String id) {skuSearchMapper.deleteById(id);}
}

controller层

@RestController
@RequestMapping(value = "/search")
public class SkuSearchController {@Autowiredprivate SkuSearchService skuSearchService;/****** 增加索引*/@PostMapping(value = "/add")public RespResult add(@RequestBody SkuEs skuEs){skuSearchService.add(skuEs);return RespResult.ok();}/**** 删除索引*/@DeleteMapping(value = "/del/{id}")public RespResult del(@PathVariable(value = "id")String id){skuSearchService.del(id);return RespResult.ok();}
}

4. mall-canal-service

基于上一篇文章Springcloud Alibaba使用Canal将Mysql数据实时同步到Redis保证缓存的一致性-CSDN博客的canal服务。

添加以下代码：

feign接口

@FeignClient(value = "mall-search")public interface SkuSearchFeign {@PostMapping(value = "/search/add")RespResult add(@RequestBody SkuEs skuEs);/**** 删除索引*/@DeleteMapping(value = "/search/del/{id}")RespResult del(@PathVariable(value = "id")String id);
}

canal设计代码如下：

@Component
@CanalTable(value = "sku")
public class SkuSearchHandler implements EntryHandler<Sku> {@AutowiredSkuSearchFeign skuSearchFeign;@Overridepublic void insert(Sku sku) {System.out.println(sku);String jsonString = JSON.toJSONString(sku);SkuEs skuEs = JSON.parseObject(jsonString, SkuEs.class);skuSearchFeign.add(skuEs);}@Overridepublic void update(Sku before, Sku after) {System.out.println(after);String jsonString = JSON.toJSONString(after);SkuEs skuEs = JSON.parseObject(jsonString, SkuEs.class);System.out.println(skuEs);skuSearchFeign.add(skuEs);}@Overridepublic void delete(Sku sku) {System.out.println(sku);skuSearchFeign.del(sku.getId());}
}