@Service
public class UserService {
    @Autowired
    private UserMapper userMapper;
    
    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;
    
    @Transactional
    public void addUser(User user) {
        // 寫入MySQL
        userMapper.insert(user);
        
        // 寫入Elasticsearch
        IndexQuery indexQuery = new IndexQueryBuilder()
            .withObject(user)
            .withId(user.getId().toString())
            .build();
        elasticsearchTemplate.index(indexQuery);
    }
    
    @Transactional
    public void updateUser(User user) {
        // 更新MySQL
        userMapper.updateById(user);
        
        // 更新Elasticsearch
        IndexRequest request = new IndexRequest("user_index")
            .id(user.getId().toString())
            .source(JSON.toJSONString(user), XContentType.JSON);
        elasticsearchTemplate.getClient().index(request, RequestOptions.DEFAULT);
    }
}

@Component
public class UserSyncTask {
    @Autowired
    private UserMapper userMapper;
    
    @Autowired
    private UserESRepository userESRepository;
    
    // 每5分鐘執(zhí)行一次
    @Scheduled(fixedRate = 5 * 60 * 1000)
    public void syncUserToES() {
        // 查詢最近更新的數(shù)據(jù)
        Date lastSyncTime = getLastSyncTime();
        List<User> updatedUsers = userMapper.selectUpdatedAfter(lastSyncTime);
        
        // 同步到ES
        for (User user : updatedUsers) {
            userESRepository.save(user);
        }
        
        // 更新最后同步時(shí)間
        updateLastSyncTime(new Date());
    }
    
    // 獲取最后同步時(shí)間
    private Date getLastSyncTime() {
        // 從數(shù)據(jù)庫(kù)或Redis中獲取
        // ...
    }
}

public class BinlogSyncService {
    public void startSync() {
        BinaryLogClient client = new BinaryLogClient("localhost", 3306, "username", "password");
        
        client.registerEventListener(new BinaryLogClient.EventListener() {
            @Override
            public void onEvent(Event event) {
                EventData eventData = event.getData();
                
                if (eventData instanceof WriteRowsEventData) {
                    // 插入操作
                    WriteRowsEventData writeData = (WriteRowsEventData) eventData;
                    processInsertEvent(writeData);
                } elseif (eventData instanceof UpdateRowsEventData) {
                    // 更新操作
                    UpdateRowsEventData updateData = (UpdateRowsEventData) eventData;
                    processUpdateEvent(updateData);
                } elseif (eventData instanceof DeleteRowsEventData) {
                    // 刪除操作
                    DeleteRowsEventData deleteData = (DeleteRowsEventData) eventData;
                    processDeleteEvent(deleteData);
                }
            }
        });
        
        client.connect();
    }
    
    private void processInsertEvent(WriteRowsEventData eventData) {
        // 處理插入事件，同步到ES
        for (Serializable[] row : eventData.getRows()) {
            User user = convertRowToUser(row);
            syncToElasticsearch(user, "insert");
        }
    }
    
    private void syncToElasticsearch(User user, String operation) {
        // 同步到ES的實(shí)現(xiàn)
        // ...
    }
}

# canal.properties 配置
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=username
canal.instance.dbPassword=password
canal.instance.connectionCharset=UTF-8
canal.instance.filter.regex=.*\..*

public class CanalClientExample {
    public static void main(String[] args) {
        // 創(chuàng)建Canal連接
        CanalConnector connector = CanalConnectors.newSingleConnector(
            new InetSocketAddress("127.0.0.1", 11111), "example", "", "");
        
        try {
            connector.connect();
            connector.subscribe(".*\..*");
            
            while (true) {
                Message message = connector.getWithoutAck(100);
                long batchId = message.getId();
                
                if (batchId != -1 && !message.getEntries().isEmpty()) {
                    processEntries(message.getEntries());
                    connector.ack(batchId); // 提交確認(rèn)
                }
                
                Thread.sleep(1000);
            }
        } finally {
            connector.disconnect();
        }
    }
    
    private static void processEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                
                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    if (rowChange.getEventType() == CanalEntry.EventType.INSERT) {
                        processInsert(rowData);
                    } elseif (rowChange.getEventType() == CanalEntry.EventType.UPDATE) {
                        processUpdate(rowData);
                    } elseif (rowChange.getEventType() == CanalEntry.EventType.DELETE) {
                        processDelete(rowData);
                    }
                }
            }
        }
    }
}

@Service
public class UserService {
    @Autowired
    private UserMapper userMapper;
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @Transactional
    public void addUser(User user) {
        // 寫入MySQL
        userMapper.insert(user);
        
        // 發(fā)送消息到MQ
        rabbitTemplate.convertAndSend("user.exchange", "user.add", user);
    }
    
    @Transactional
    public void updateUser(User user) {
        // 更新MySQL
        userMapper.updateById(user);
        
        // 發(fā)送消息到MQ
        rabbitTemplate.convertAndSend("user.exchange", "user.update", user);
    }
}

@Component
public class UserMQConsumer {
    @Autowired
    private UserESRepository userESRepository;
    
    @RabbitListener(queues = "user.queue")
    public void processUserAdd(User user) {
        userESRepository.save(user);
    }
    
    @RabbitListener(queues = "user.queue")
    public void processUserUpdate(User user) {
        userESRepository.save(user);
    }
    
    @RabbitListener(queues = "user.queue")
    public void processUserDelete(Long userId) {
        userESRepository.deleteById(userId);
    }
}