整理 Canal 监听 MySQL Binlog 的缓存失效流程，并开始分析听书项目订单模块和结算入口。
本篇要点
- 理解 Binlog 与 Canal 的关系
- 梳理变更后的缓存失效
- 进入订单模块的需求分析
**学习提示：**Canal 相关示例账号已占位；这里的处理重点是识别数据变更并删除相应缓存，并非直接同步完整 Redis 对象。
内容回顾
1、专辑缓存+分布式锁
- 使用分布式锁解决缓存击穿问题
- 一、使用Redis的setnx + 过期时间 + uuid + lua脚本
- 二、Redisson框架使用
- Redisson实现分布式锁有两种机制
– 重试机制
– 看门狗机制
2、自定义注解+AOP实现缓存
3、布隆过滤器
解决缓存穿透问题
存储0或者1数组
使用多个映射函数计算数据在数组位置
存在误判率
4、mysql和redis数据一致性
– 删除缓存
– 修改数据库
– 休眠一会
– 再删除缓存
– 相当于mysql从机，监控mysql主机里面二进制日志变化
今天内容
1、数据同步工具canal
作用
底层原理
- mysql有主从复制机制，主机里面写操作，同步到从机里面，读在从机里面进行读操作
- canal相当于mysql从机的角色
- 流程
1 在mysql主机里面有二进制日志，二进制日志记录主机变化的数据
2 canal实时监控主机里面二进制日志，当日志文件有数据变化，canal读取变化数据
3 canal根据变化数据操作redis
canal使用
第一，安装canal服务
操作mysql数据库
1
| SHOW VARIABLES LIKE '%log_bin%';
|
- 为了操作方便，可以在mysql创建用户，专门做同步使用的
1
2
3
4
5
6
7
8
| #创建用户
create user canal@'%' IDENTIFIED by 'canal';
#给用户授权
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT,SUPER ON *.* TO 'canal'@'%';
#如果是MySQL8.X以上需要对加密方式进行设置
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY '<CANAL_PASSWORD>';
#刷新生效
FLUSH PRIVILEGES;
|
使用docker安装canal
1
2
3
4
5
6
7
8
9
10
| docker run -p 11111:11111 --name canal \
-e canal.destinations=tingshuTopic \
-e canal.instance.master.address=192.168.200.130:3306 \
-e canal.instance.dbUsername=canal \
-e canal.instance.dbPassword=<CANAL_PASSWORD> \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false \
-e canal.instance.filter.regex=.*\\..* \
-d canal/canal-server:v1.1.5
|
第二，SpringBoot整合canal
第一步 引入依赖
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
| <?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">
<modelVersion>4.0.0</modelVersion>
<groupId>com.atguigu</groupId>
<artifactId>service-cdc</artifactId>
<version>1.0-SNAPSHOT</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.6.RELEASE</version>
<relativePath/>
</parent>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>top.javatool</groupId>
<artifactId>canal-spring-boot-starter</artifactId>
<version>1.2.1-RELEASE</version>
</dependency>
<dependency>
<groupId>javax.persistence</groupId>
<artifactId>persistence-api</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
</dependencies>
</project>
|
第二步 创建实体类
1
2
3
4
5
6
| @Data
public class CDCEntity {
@Column(name = "id")
private Long id;
}
|
第三步 创建类实现canal里面的接口EntryHandler
第四步 实现EntryHandler接口方法实现同步逻辑
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
| @Slf4j
@Component
@CanalTable("album_info")
public class AlbumInfoCdcHandler implements EntryHandler<CDCEntity> {
@Autowired
private RedisTemplate redisTemplate;
@Override
public void insert(CDCEntity cdcEntity) {
System.out.println(cdcEntity.getId());
}
@Override
public void update(CDCEntity before, CDCEntity after) {
log.info("监听到数据修改,ID:{}", after.getId());
String key = "album:info:" + after.getId();
redisTemplate.delete(key);
}
@Override
public void delete(CDCEntity cdcEntity) {
log.info("监听到数据删除,ID:{}", cdcEntity.getId());
String key = "album:info:" + cdcEntity.getId();
redisTemplate.delete(key);
}
}
|
第五步 创建配置文件
1
2
3
4
5
6
| server:
port: 7080
canal:
destination: tingshuTopic1
server: 192.168.200.130:11111
|
可能遇到两个问题
第一个问题：因为canal不支持jdk17的，支持jdk1.8，修改idea环境jdk1.8环境
方式一：
方式二：
第二个问题：启动springBoot项目监控不到mysql日志
重启canal服务和SpringBoot项目：docker restart canal
如果还不行，把mysql服务重启
2、订单模块
订单模块需求
- 购买订单包含三部分：购买vip、购买专辑、购买声音
购买vip
- 开通vip，最终：在user_info表里面，修改is_vip=1，添加vip过期时间
购买专辑
- 首先 专辑不是免费的pay_type不是0101
- 其次 购买专辑price_type是0202
- 第三 购买专辑有折扣，普通用户使用普通折扣，如果vip用户使用vip折扣
- 购买专辑成功之后，最终user_paid_album添加一条记录
购买声音
- 专辑类型不是免费的，专辑price_type值0201
- 使用账户余额购买声音
- 声音购买没有折扣
获取vip服务配置信息
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| @Tag(name = "VIP服务配置管理接口")
@RestController
@RequestMapping("api/user/vipServiceConfig")
@SuppressWarnings({"all"})
public class VipServiceConfigApiController {
@Autowired
private VipServiceConfigService vipServiceConfigService;
@Operation(summary = "获取全部VIP 服务配置信息")
@GetMapping("findAll")
public Result<List<VipServiceConfig>> findAll(){
List<VipServiceConfig> list = this.vipServiceConfigService.list();
return Result.ok(list);
}
}
|
跳转（确认）结算页面接口
远程调用接口：用户是否购买专辑
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| @Tag(name = "用户管理接口")
@RestController
@RequestMapping("api/user/userInfo")
@SuppressWarnings({"all"})
public class UserInfoApiController {
@Autowired
private UserInfoService userInfoService;
@GuiguLogin
@GetMapping("isPaidAlbum/{albumId}")
public Result<Boolean> isPaidAlbum(@PathVariable("albumId") Long albumId) {
Long userId = AuthContextHolder.getUserId();
Boolean isPaid = userInfoService.isPaidAlbum(albumId,userId);
return Result.ok(isPaid);
}
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
@Override
public Boolean isPaidAlbum(Long albumId, Long userId) {
LambdaQueryWrapper<UserPaidAlbum> wrapper01 = new LambdaQueryWrapper<>();
wrapper01.eq(UserPaidAlbum::getAlbumId, albumId);
wrapper01.eq(UserPaidAlbum::getUserId, userId);
Long count = userPaidAlbumMapper.selectCount(wrapper01);
return count>0;
}
|
远程调用接口：获取vip信息
1
2
3
4
5
6
7
8
9
10
|
@Operation(summary = "根据id获取VIP服务配置信息")
@GetMapping("getVipServiceConfig/{id}")
public Result<VipServiceConfig> getVipServiceConfig(@PathVariable Long id) {
return Result.ok(vipServiceConfigService.getById(id));
}
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| @Tag(name = "订单管理")
@RestController
@RequestMapping("api/order/orderInfo")
@SuppressWarnings({"all"})
public class OrderInfoApiController {
@Autowired
private OrderInfoService orderInfoService;
@GuiguLogin
@Operation(summary = "确认订单")
@PostMapping("trade")
public Result<OrderInfoVo> trade(@RequestBody @Validated TradeVo tradeVo) {
Long userId = AuthContextHolder.getUserId();
OrderInfoVo orderInfoVo = orderInfoService.trade(tradeVo,userId);
return Result.ok(orderInfoVo);
}
}
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
@Override
public OrderInfoVo trade(TradeVo tradeVo, Long userId) {
Result<UserInfoVo> resultUserInfoVo =
userInfoFeignClient.getUserInfoVo(userId);
UserInfoVo userInfoVo = resultUserInfoVo.getData();
Assert.notNull(userInfoVo,"用户信息为空");
BigDecimal originalAmount = new BigDecimal("0.00");
BigDecimal derateAmount = new BigDecimal("0.00");
BigDecimal orderAmount = new BigDecimal("0.00");
List<OrderDetailVo> orderDetailVoList = new ArrayList<>();
List<OrderDerateVo> orderDerateVoList = new ArrayList<>();
String itemType = tradeVo.getItemType();
if("1001".equals(itemType)) {
Result<Boolean> resultUserPaid =
userInfoFeignClient.isPaidAlbum(tradeVo.getItemId());
Boolean isPaid = resultUserPaid.getData();
if(isPaid) {
throw new GuiguException(300,"专辑购买过");
}
Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(tradeVo.getItemId());
AlbumInfo albumInfo = albumInfoResult.getData();
Assert.notNull(albumInfo,"专辑数据为空");
Integer isVip = userInfoVo.getIsVip();
if(isVip.intValue()==0) {
originalAmount = albumInfo.getPrice();
BigDecimal discount = albumInfo.getDiscount();
if(discount.intValue()!=-1) {
derateAmount = originalAmount.multiply(
new BigDecimal("10").subtract(discount))
.divide(new BigDecimal(10),2,RoundingMode.HALF_UP);
}
orderAmount = originalAmount.subtract(derateAmount);
} else if(isVip.intValue()==1 && userInfoVo.getVipExpireTime().before(new Date())) {
originalAmount = albumInfo.getPrice();
BigDecimal discount = albumInfo.getDiscount();
if(discount.intValue()!=-1) {
derateAmount = originalAmount.multiply(
new BigDecimal("10").subtract(discount))
.divide(new BigDecimal(10),2,RoundingMode.HALF_UP);
}
orderAmount = originalAmount.subtract(derateAmount);
} else {
BigDecimal vipDiscount = albumInfo.getVipDiscount();
originalAmount = albumInfo.getPrice();
if(vipDiscount.intValue()!=-1) {
derateAmount = originalAmount.multiply(
new BigDecimal("10").subtract(vipDiscount))
.divide(new BigDecimal(10),2,RoundingMode.HALF_UP);
}
orderAmount = originalAmount.subtract(derateAmount);
}
OrderDetailVo orderDetailVo = new OrderDetailVo();
orderDetailVo.setItemId(tradeVo.getItemId());
orderDetailVo.setItemName(albumInfo.getAlbumTitle());
orderDetailVo.setItemUrl(albumInfo.getCoverUrl());
orderDetailVo.setItemPrice(albumInfo.getPrice());
orderDetailVoList.add(orderDetailVo);
if (originalAmount.subtract(orderAmount).doubleValue() != 0) {
OrderDerateVo orderDerateVo = new OrderDerateVo();
orderDerateVo.setDerateType(SystemConstant.ORDER_DERATE_ALBUM_DISCOUNT);
orderDerateVo.setDerateAmount(originalAmount.subtract(orderAmount));
orderDerateVoList.add(orderDerateVo);
}
} else if("1003".equals(itemType)) {
Result<VipServiceConfig> vipServiceConfigResult =
vipServiceConfigFeignClient.getVipServiceConfig(tradeVo.getItemId());
VipServiceConfig vipServiceConfig = vipServiceConfigResult.getData();
Assert.notNull(vipServiceConfig,"返回vip配置对象不能为空");
originalAmount = vipServiceConfig.getPrice();
orderAmount = vipServiceConfig.getDiscountPrice();
derateAmount = originalAmount.subtract(orderAmount);
OrderDetailVo orderDetailVo = new OrderDetailVo();
orderDetailVo.setItemId(tradeVo.getItemId());
orderDetailVo.setItemName("VIP会员"+vipServiceConfig.getName());
orderDetailVo.setItemUrl(vipServiceConfig.getImageUrl());
orderDetailVo.setItemPrice(vipServiceConfig.getDiscountPrice());
orderDetailVoList.add(orderDetailVo);
if (originalAmount.subtract(orderAmount).doubleValue() != 0) {
OrderDerateVo orderDerateVo = new OrderDerateVo();
orderDerateVo.setDerateType(SystemConstant.ORDER_DERATE_VIP_SERVICE_DISCOUNT);
orderDerateVo.setDerateAmount(originalAmount.subtract(orderAmount));
orderDerateVoList.add(orderDerateVo);
}
} else if("1002".equals(itemType)) {
}
OrderInfoVo orderInfoVo = new OrderInfoVo();
orderInfoVo.setItemType(tradeVo.getItemType());
orderInfoVo.setOriginalAmount(originalAmount);
orderInfoVo.setDerateAmount(derateAmount);
orderInfoVo.setOrderAmount(orderAmount);
orderInfoVo.setOrderDetailVoList(orderDetailVoList);
orderInfoVo.setOrderDerateVoList(orderDerateVoList);
orderInfoVo.setTimestamp(SignHelper.getTimestamp());
orderInfoVo.setPayWay(SystemConstant.ORDER_PAY_WAY_WEIXIN);
String key = "user:trade:" + userId;
String value = UUID.randomUUID().toString().replaceAll("-", "");
redisTemplate.opsForValue().set(key,value);
orderInfoVo.setTradeNo(value);
String jsonString = JSON.toJSONString(orderInfoVo);
Map<String, Object> parameterMap = JSON.parseObject(jsonString, Map.class);
String sign = SignHelper.getSign(parameterMap);
orderInfoVo.setSign(sign);
return orderInfoVo;
}
|
防止重复提交
防止数据被修改
