2020-12-11
19587 ℃
原创
Canal
1. canal同步广告
canal可以用来监控数据库数据的变化，从而获得新增数据，或者修改的数据。
canal是应阿里巴巴存在杭州和美国的双机房部署，存在跨机房同步的业务需求而提出的。
阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务。
1.1 Canal工作原理
原理相对比较简单：
- canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议
- mysql master收到dump请求，开始推送binary log给slave(也就是canal)
- canal解析binary log对象(原始为byte流)
canal需要使用到mysql，我们需要先安装mysql,但canal是基于mysql的主从模式实现的，所以必须先开启binlog.
1.2 开启binlog模式
先使用docker 创建mysql容器,此处不再演示.
(1) 连接到mysql中,并修改/etc/mysql/mysql.conf.d/mysqld.cnf 需要开启主 从模式，开启binlog模式。
执行如下命令，编辑mysql配置文件
命令行如下：
docker exec -it mysql /bin/bash
cd /etc/mysql/mysql.conf.d
vi mysqld.cnf
修改mysqld.cnf配置文件，添加如下配置：
上图配置如下：
log-bin/var/lib/mysql/mysql-bin
server-id=12345
(2) 创建账号 用于测试使用,
使用root账号创建用户并授予权限
create user canal@'%' IDENTIFIED by 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;
(3)重启mysql容器
docker restart mysql
1.3 canal容器安装
下载镜像：
docker pull docker.io/canal/canal-server
容器安装
docker run -p 11111:11111 --name canal -d docker.io/canal/canal-server
进入容器,修改核心配置canal.properties 和instance.properties，canal.properties 是canal自身的配置，instance.properties是需要同步数据的数据库连接配置。
执行代码如下:
docker exec -it canal /bin/bash
cd canal-server/conf/
vi canal.properties
cd example/
vi instance.properties
修改canal.properties的id，不能和mysql的server-id重复，如下图：
修改instance.properties,配置数据库连接地址:
这里的canal.instance.filter.regex有多种配置，如下：
可以参考地址如下:
https://github.com/alibaba/canal/wiki/AdminGuide
mysql 数据解析关注的表，Perl正则表达式.
多个正则之间以逗号(,)分隔，转义符需要双斜杠(\\)
常见例子：
1. 所有表：.* or .*\\..*
2. canal schema下所有表： canal\\..*
3. canal下的以canal打头的表：canal\\.canal.*
4. canal schema下的一张表：canal.test1
5. 多个规则组合使用：canal\\..*,mysql.test1,mysql.test2 (逗号分隔)
注意：此过滤条件只针对row模式的数据有效(ps. mixed/statement因为不解析sql，所以无法准确提取tableName进行过滤)
配置完成后，设置开机启动，并记得重启canal。
docker update --restart=always canal
docker restart canal
1.4 canal微服务搭建
当用户执行 数据库的操作的时候，binlog 日志会被canal捕获到，并解析出数据。我们就可以将解析出来的数据进行同步到redis中即可。
思路：创建一个独立的程序，并监控canal服务器，获取binlog日志，解析数据，将数据更新到redis中。这样广告的数据就更新了。
(1)安装辅助jar包
在canal\spring-boot-starter-canal-master中有一个工程starter-canal，它主要提供了SpringBoot环境下canal的支持，我们需要先安装该工程，在starter-canal目录下执行mvn install，如下图：
(2)canal微服务工程搭建
在changgou-service下创建changgou-service-canal工程，并引入相关配置。
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>changgou-service</artifactId>
<groupId>com.changgou</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>changgou-service-canal</artifactId>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!--canal依赖-->
<dependency>
<groupId>com.xpand</groupId>
<artifactId>starter-canal</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
</dependencies>
</project>
application.yml配置
server:
port: 18082
spring:
application:
name: canal
eureka:
client:
service-url:
defaultZone: http://127.0.0.1:7001/eureka
instance:
prefer-ip-address: true
feign:
hystrix:
enabled: true
#hystrix 配置
hystrix:
command:
default:
execution:
timeout:
#如果enabled设置为false，则请求超时交给ribbon控制
enabled: true
isolation:
strategy: SEMAPHORE
#canal配置
canal:
client:
instances:
example:
host: 192.168.211.132
port: 11111
(3)监听创建
创建一个CanalDataEventListener类，实现对表增删改操作的监听，代码如下：
package com.changgou.canal.listener;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.xpand.starter.canal.annotation.*;
@CanalEventListener
public class CanalDataEventListener {
/***
* 增加数据监听
* @param eventType 当前操作类型
* @param rowData 发生变更的一行数据
*/
@InsertListenPoint
public void onEventInsert(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
System.out.println("=========================新增=====================");
rowData.getAfterColumnsList().forEach((c) -> System.out.println("列名: " + c.getName() + " ,新增的数据 " + c.getValue()));
}
/***
* 修改数据监听
* @param rowData
*/
@UpdateListenPoint
public void onEventUpdate(CanalEntry.RowData rowData) {
System.out.println("=======================修改==================");
rowData.getAfterColumnsList().forEach((c) -> System.out.println("列名: " + c.getName() + " ,变更的数据 " + c.getValue()));
}
/***
* 删除数据监听
* @param eventType
*/
@DeleteListenPoint
public void onEventDelete(CanalEntry.EventType eventType,CanalEntry.RowData rowData) {
System.out.println("====================删除======================");
rowData.getBeforeColumnsList().forEach((c) -> System.out.println("列名: " + c.getName() + " ,删除的数据 " + c.getValue()));
}
/***
* 自定义数据修改监听
* @param eventType
* @param rowData
*/
@ListenPoint(
destination = "example",//指定实例的地址
schema = "changgou_content",//指定监控的数据
table = {"tb_content_category", "tb_content"},//指定监控的表
eventType = CanalEntry.EventType.UPDATE)//指定监听类型
public void onEventCustomUpdate(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
System.err.println("DeleteListenPoint");
rowData.getAfterColumnsList().forEach((c) -> System.out.println("By--Annotation: " + c.getName() + " :: " + c.getValue()));
}
}
(4)启动类创建
在com.changgou中创建启动类，代码如下：
@SpringBootApplication(exclude={DataSourceAutoConfiguration.class})
@EnableEurekaClient
@EnableCanalClient
public class CanalApplication {
public static void main(String[] args) {
SpringApplication.run(CanalApplication.class,args);
}
}
(5)测试
1.5 广告同步
如上图，每次执行广告操作的时候，会记录操作日志到，然后将操作日志发送给canal，canal将操作记录发送给canal微服务，canal微服务根据修改的分类ID调用content微服务查询分类对应的所有广告，canal微服务再将所有广告存入到Redis缓存。
6.5.1 content微服务搭建
在changgou-service中搭建changgou-service-content微服务，对应的dao、service、controller、pojo由代码生成器生成。
首先在changgou-service-api中创建changgou-service-content-api,将pojo拷贝到API工程中，如下图：
(1)pom.xml配置
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<parent>
<artifactId>changgou-service</artifactId>
<groupId>com.changgou</groupId>
<version>1.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<artifactId>changgou-service-content</artifactId>
<dependencies>
<dependency>
<groupId>com.changgou</groupId>
<artifactId>changgou-common</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.changgou</groupId>
<artifactId>changgou-service-content-api</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
</dependencies>
</project>
(2)application.yml配置
server:
port: 18084
spring:
application:
name: content
datasource:
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://192.168.211.132:3306/changgou_content?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC
username: root
password: 123456
eureka:
client:
service-url:
defaultZone: http://127.0.0.1:7001/eureka
instance:
prefer-ip-address: true
feign:
hystrix:
enabled: true
mybatis:
configuration:
map-underscore-to-camel-case: true #开启驼峰功能
#hystrix 配置
hystrix:
command:
default:
execution:
timeout:
#如果enabled设置为false，则请求超时交给ribbon控制
enabled: true
isolation:
strategy: SEMAPHORE
(3)启动类创建
@SpringBootApplication
@EnableEurekaClient
@MapperScan(basePackages = {"com.changgou.content.dao"})
public class ContentApplication {
public static void main(String[] args) {
SpringApplication.run(ContentApplication.class);
}
}
6.5.2 广告查询
在content微服务中，添加根据分类查询广告。
(1)业务层
修改changgou-service-content的com.changgou.content.service.ContentService接口，添加根据分类ID查询广告数据，代码如下：
/***
* 根据categoryId查询广告集合
* @param id
* @return
*/
List<Content> findByCategory(Long id);
修改changgou-service-content的com.changgou.content.service.impl.ContentServiceImpl接口实现类，添加根据分类ID查询广告数据，代码如下：
/***
* 根据分类ID查询
* @param id
* @return
*/
@Override
public List<Content> findByCategory(Long id) {
Content content = new Content();
content.setCategoryId(id);
content.setStatus("1");
return contentMapper.select(content);
}
(2)控制层
修改changgou-service-content的com.changgou.content.controller.ContentController,添加根据分类ID查询广告数据，代码如下：
/***
* 根据categoryId查询广告集合
*/
@GetMapping(value = "/list/category/{id}")
public Result<List<Content>> findByCategory(@PathVariable Long id){
//根据分类ID查询广告集合
List<Content> contents = contentService.findByCategory(id);
return new Result<List<Content>>(true,StatusCode.OK,"查询成功！",contents);
}
(3)feign配置
在changgou-service-content-api工程中添加feign，代码如下：
@FeignClient(name="content")
@RequestMapping(value = "/content")
public interface ContentFeign {
/***
* 根据分类ID查询所有广告
*/
@GetMapping(value = "/list/category/{id}")
Result<List<Content>> findByCategory(@PathVariable Long id);
}
6.5.3 同步实现
在canal微服务中修改如下:
(1)配置redis
修改application.yml配置文件，添加redis配置，如下代码：
(2)启动类中开启feign
修改CanalApplication，添加@EnableFeignClients注解，代码如下：
(3)同步实现
修改监听类CanalDataEventListener，实现监听广告的增删改，并根据增删改的数据使用feign查询对应分类的所有广告，将广告存入到Redis中，代码如下：
上图代码如下：
@CanalEventListener
public class CanalDataEventListener {
@Autowired
private ContentFeign contentFeign;
//字符串
@Autowired
private StringRedisTemplate stringRedisTemplate;
//自定义数据库的 操作来监听
//destination = "example"
@ListenPoint(destination = "example",
schema = "changgou_content",
table = {"tb_content", "tb_content_category"},
eventType = {
CanalEntry.EventType.UPDATE,
CanalEntry.EventType.DELETE,
CanalEntry.EventType.INSERT})
public void onEventCustomUpdate(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
//1.获取列名 为category_id的值
String categoryId = getColumnValue(eventType, rowData);
//2.调用feign 获取该分类下的所有的广告集合
Result<List<Content>> categoryresut = contentFeign.findByCategory(Long.valueOf(categoryId));
List<Content> data = categoryresut.getData();
//3.使用redisTemplate存储到redis中
stringRedisTemplate.boundValueOps("content_" + categoryId).set(JSON.toJSONString(data));
}
private String getColumnValue(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
String categoryId = "";
//判断 如果是删除 则获取beforlist
if (eventType == CanalEntry.EventType.DELETE) {
for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
if (column.getName().equalsIgnoreCase("category_id")) {
categoryId = column.getValue();
return categoryId;
}
}
} else {
//判断 如果是添加 或者是更新 获取afterlist
for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
if (column.getName().equalsIgnoreCase("category_id")) {
categoryId = column.getValue();
return categoryId;
}
}
}
return categoryId;
}
}
测试：
修改数据库数据，可以看到Redis中的缓存跟着一起变化
评论