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 {

    @Id
    private 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-search
  cloud:
    nacos:
      config:
        file-extension: yaml
        server-addr: localhost:8848
      discovery:
        #Nacos的注册地址
        server-addr: localhost:8848
  #Elasticsearch服务配置 6.8.12
  elasticsearch:
      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 {

    @Autowired
    SkuSearchMapper skuSearchMapper;

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    @Override
    public void add(SkuEs skuEs) {
        skuSearchMapper.save(skuEs);
    }

    @Override
    public void del(String id) {
       skuSearchMapper.deleteById(id);
    }
}

controller层

@RestController
@RequestMapping(value = "/search")
public class SkuSearchController {

    @Autowired
    private 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> {

    @Autowired
    SkuSearchFeign skuSearchFeign;

    @Override
    public void insert(Sku sku) {
        System.out.println(sku);
        String jsonString = JSON.toJSONString(sku);
        SkuEs skuEs = JSON.parseObject(jsonString, SkuEs.class);
         skuSearchFeign.add(skuEs);
    }

    @Override
    public 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);
    }

    @Override
    public void delete(Sku sku) {
        System.out.println(sku);
        skuSearchFeign.del(sku.getId());
    }
}