Canal实现MySql和redis的数据同步 - 黑帽联盟

server-id=140 # 给mysql服务创建一个唯一id
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式

service mysql restart

canal.instance.master.address=10.70.20.140:3306       #mysql ip+端口
canal.instance.dbUsername=root                      #mysql 账号
canal.instance.dbPassword=123456                      #mysql 密码
canal.instance.filter.regex=canaldb.canaltable       #需要同步的库、表

# bin目录下启动canal
startup.sh

<dependencies>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-web</artifactId>
      </dependency>

      <dependency>
         <groupId>org.projectlombok</groupId>
         <artifactId>lombok</artifactId>
         <optional>true</optional>
      </dependency>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-test</artifactId>
         <scope>test</scope>
      </dependency>
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-data-redis</artifactId>
      </dependency>
      
      <dependency>
         <groupId>com.alibaba</groupId>
         <artifactId>druid-spring-boot-starter</artifactId>
         <version>1.1.17</version>
      </dependency>
      
      <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-jdbc</artifactId>
      </dependency>
      
      <dependency>
         <groupId>mysql</groupId>
         <artifactId>mysql-connector-java</artifactId>
         <version>8.0.29</version>
      </dependency>
</dependencies>

canal:
  server: 10.70.20.140:11111
  destination: example
spring:
  redis:
host: 10.70.20.140
password: 123456
port: 6379

  datasource:
druid:
   driver-class-name: com.mysql.cj.jdbc.Driver
   url: jdbc:mysql://10.70.20.140:3306/canaldb
   username: root
   password: fe781212

@Configuration
public class RedisConfig {

@Resource
private RedisConnectionFactory factory;

//使用JSON进行序列化
@Bean
public RedisTemplate<Object, Object> redisTemplate() {
      RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();

      redisTemplate.setConnectionFactory(factory);
      //JSON格式序列化
      GenericFastJsonRedisSerializer serializer = new GenericFastJsonRedisSerializer();
      //key的序列化
      redisTemplate.setKeySerializer(serializer);
      //value的序列化
      redisTemplate.setValueSerializer(serializer);
      //hash结构key的虚拟化
      redisTemplate.setHashKeySerializer(new StringRedisSerializer());
      //hash结构value的虚拟化
      redisTemplate.setHashValueSerializer(serializer);
      return redisTemplate;
}

}

@CanalTable("canaltable")
@Component
@Slf4j
public class UserHandler implements EntryHandler<User> {
      @Autowired
      private RedisTemplate<Object,Object> redisTemplate;

@Override
public void insert(User user) {
      redisTemplate.opsForValue().set("USER:"+user.getId(),user);
}

@Override
public void update(User before, User after) {
      redisTemplate.opsForValue().set("USER："+before.getId(),after);
}

@Override
public void delete(User user) {
      redisTemplate.delete("USER:"+user.getId());
}
}