server-id=1	                     #master端的ID號【必須是唯一的】；
log-bin=D:\MySQL\binlog\mysql-bin.log	  #同步的日志路徑，一定注意這個目錄要是mysql有權(quán)限寫入的
binlog_format=row	               #行級，記錄每次操作后每行記錄的變化。
binlog-do-db=db_xiaomi	           #指定庫，縮小監(jiān)控的范圍。

        <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.指定要監(jiān)控的數(shù)據(jù)庫
        canalConnector.subscribe("db.xiaomi.*");
        //4.獲取 Message
        Message message = canalConnector.get(100);
        List<CanalEntry.Entry> entries = message.getEntries();
        if (entries.size() <= 0) {
            System.out.println("沒有數(shù)據(jù)，休息一會");
            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)) {
                    //  序列化數(shù)據(jù)
                    ByteString storeValue = entry.getStoreValue();
                    //  反序列化
                    CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                    // 獲取事件類型
                    CanalEntry.EventType eventType = rowChange.getEventType();
                    // 獲取具體的數(shù)據(jù)
                    List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
                    // 遍歷并打印數(shù)據(jù)
                    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監(jiān)聽類
 * @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(">>>>>>>應(yīng)該是單獨起一個線程<<<<<<<<<<");
        
    }
}

/** 自定義線程池 bean 用于Async異步調(diào)用
 * @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();
        // 設(shè)置核心線程數(shù)
        executor.setCorePoolSize(8);
        // 設(shè)置最大線程數(shù)
        executor.setMaxPoolSize(20);
        // 設(shè)置隊列大小
        executor.setQueueCapacity(Integer.MAX_VALUE);
        // 設(shè)置線程活躍時間(秒)
        executor.setKeepAliveSeconds(60);
        // 設(shè)置線程名前綴+分組名稱
        executor.setThreadNamePrefix("AsyncOperationThread-");
        executor.setThreadGroupName("AsyncOperationGroup");
        // 所有任務(wù)結(jié)束后關(guān)閉線程池
        executor.setWaitForTasksToCompleteOnShutdown(true);
        // 初始化
        executor.initialize();
        return executor;
    }

}

/**異步數(shù)據(jù)更新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數(shù)據(jù)更新消費者監(jiān)聽方法
     */
    @RabbitListener(queues = MqConstants.QUEUE_NAME)
    public void updateRedisDataByAsync(Map<String,Object> msg){
        System.out.println("監(jiān)聽到數(shù)據(jù)變化：");
        System.out.println("數(shù)據(jù)變化商品id:"+msg.get("id"));  //正常情況Redis應(yīng)該每個商品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"));
    }

}