Redis和数据库的一致性(Canal+MQ) 的实现

server-id=1	                     #master端的ID号【必须是唯一的】；
log-bin=D:\MySQL\binlog\mysql-bin.log	  #同步的日志路径，一定注意这个目录要是mysql有权限写入的
binlog_format=row	               #行级，记录每次操作后每行记录的变化。
binlog-do-db=db_xiaomi	           #指定库，缩小监控的范围。

        <dependency>
            <groupId>com.alibaba.otter</groupId>
            <artifactId>canal.client</artifactId>
            <version>1.1.0</version>
        </dependency>

public static void main(String[] args) throws InvalidProtocolBufferException {
    CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("localhost", 11111), "example", "", "");
    while (true) {
        //2.获取连接
        canalConnector.connect();
        //3.指定要监控的数据库
        canalConnector.subscribe("db.xiaomi.*");
        //4.获取 Message
        Message message = canalConnector.get(100);
        List<CanalEntry.Entry> entries = message.getEntries();
        if (entries.size() <= 0) {
            System.out.println("没有数据，休息一会");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        } else {
            for (CanalEntry.Entry entry : entries) {
                // 获取表名
                String tableName = entry.getHeader().getTableName();
                //  Entry 类型
                CanalEntry.EntryType entryType = entry.getEntryType();
                //  判断 entryType 是否为 ROWDATA
                if (CanalEntry.EntryType.ROWDATA.equals(entryType)) {
                    //  序列化数据
                    ByteString storeValue = entry.getStoreValue();
                    //  反序列化
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                    // 获取事件类型
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    // 获取具体的数据
                    List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                    // 遍历并打印数据
                    for (CanalEntry.RowData rowData : rowDatasList) {
                        List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
                        Map<String, Object> bMap = new HashMap<>();
                        for (CanalEntry.Column column : beforeColumnsList) {
                            bMap.put(column.getName(), column.getValue());
                        }
                        Map<String, Object> afMap = new HashMap<>();
                        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
                        for (CanalEntry.Column column : afterColumnsList) {
                            afMap.put(column.getName(), column.getValue());
                        }
                        System.out.println("表名:" + tableName + ",操作类型:" + eventType);
                        System.out.println("改前:" + bMap);
                        System.out.println("改后:" + afMap);
                    }
                }
            }
        }
    }
}

/**Canal监听类
 * @author 12547
 * @version 1.0
 * @Date 2024/3/19 20:44
 */
@Component
public class CanalRunner implements CommandLineRunner {

  
    @Override
    public void run(String... args) throws Exception {
        System.out.println(">>>>>>>此处并不适用Canal的运行方式<<<<<<<<<<");
        System.out.println(">>>>>>>应该是单独起一个线程<<<<<<<<<<");
        
    }
}

/** 自定义线程池 bean 用于Async异步调用
 * @author 12547
 * @version 1.0
 * @Date 2024/3/20 15:45
 */
@Configuration
@EnableAsync
public class AsyncConfig {

    /**
     * 自定义线程
     * @return
     */
    @Bean("asyncPoll")
    public ThreadPoolTaskExecutor asyncOperationExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        // 设置核心线程数
        executor.setCorePoolSize(8);
        // 设置最大线程数
        executor.setMaxPoolSize(20);
        // 设置队列大小
        executor.setQueueCapacity(Integer.MAX_VALUE);
        // 设置线程活跃时间(秒)
        executor.setKeepAliveSeconds(60);
        // 设置线程名前缀+分组名称
        executor.setThreadNamePrefix("AsyncOperationThread-");
        executor.setThreadGroupName("AsyncOperationGroup");
        // 所有任务结束后关闭线程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 初始化
        executor.initialize();
        return executor;
    }

}

/**异步数据更新Redis类
 * @author 12547
 * @version 1.0
 * @Date 2024/3/20 15:49
 */
@Component
public class RedisDataListenerService {

    @Autowired
    private CacheService cacheService;

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * Redis数据更新消费者监听方法
     */
    @RabbitListener(queues = MqConstants.QUEUE_NAME)
    public void updateRedisDataByAsync(Map<String,Object> msg){
        System.out.println("监听到数据变化：");
        System.out.println("数据变化商品id:"+msg.get("id"));  //正常情况Redis应该每个商品id一个key  TODO 需要改造详情缓存查询将List<phone>改为单独的一个phone对象
        redisTemplate.opsForHash().putAll(msg.get("id").toString(),msg);
        System.out.println(msg.get("id").toString());
        System.out.println(cacheService.getHashCache(msg.get("id").toString(), "num"));
    }

}