如何在Neo4j与PostgreSQL间实现高效数据同步

(:Teacher {id: "T1", name: "张老师", subject: "数学"})-[:TEACHES]->(:Class {id: "C1", name: "一年级"})

(:Class {id: "C1", name: "一年级"})-[:HAS_STUDENT]->(:Student {id: "S1", name: "李明", age: 12})

@Configuration
public class TransactionConfig {
    @Autowired
    private DataSource dataSource;
    @Autowired
    private Driver neo4jDriver;
    @Bean("postgresTransactionManager")
    public PlatformTransactionManager postgresTransactionManager() {
        return new DataSourceTransactionManager(dataSource);
    }
    @Bean("neo4jTransactionManager")
    public PlatformTransactionManager neo4jTransactionManager() {
        return new Neo4jTransactionManager(neo4jDriver);
    }
    @Bean("transactionManager")
    public PlatformTransactionManager chainedTransactionManager(
            @Qualifier("postgresTransactionManager") PlatformTransactionManager postgresTransactionManager,
            @Qualifier("neo4jTransactionManager") PlatformTransactionManager neo4jTransactionManager) {
        return new ChainedTransactionManager(
                postgresTransactionManager,
                neo4jTransactionManager
        );
    }
}

    public void processChangeLog(String changeLog) {
        try {
            String operation = extractOperation(changeLog);
            String table = extractTable(changeLog);
            String id = extractId(changeLog);
            log.debug("Processing change: operation={}, table={}, id={}", operation, table, id);
            switch (table) {
                case "public.xfc_teacher":
                    processTeacherChange(operation, id, changeLog);
                    break;
                case "public.xfc_class":
                    processClassChange(operation, id, changeLog);
                    break;
                case "public.xfc_student":
                    processStudentChange(operation, id, changeLog);
                    break;
                case "public.xfc_brid_teacher_and_class":
                    processTeacherClassRelation(operation, changeLog);
                    break;
                default:
                    log.warn("未知的表操作: {}", table);
            }
        } catch (Exception e) {
            log.error("处理变更日志时发生错误: {}", changeLog, e);
        }
    }

public class DatabaseChangeService {
    private final JdbcTemplate jdbcTemplate;
    private final ChangeLogProcessor changeLogProcessor;
    private static final String SLOT_NAME = "neo4j_replication_slot";
    private static final long POLL_INTERVAL = 1000; // 1秒
    private static final int MAX_RETRIES = 3;
    private volatile boolean running = true;
    @PostConstruct
    public void startListening() {
        new Thread(this::initializeReplicationSlot, "ReplicationListener").start();
    }
    private void initializeReplicationSlot() {
        try {
            if (!isSlotExists(SLOT_NAME)) {
                createReplicationSlot(SLOT_NAME);
                log.info("Created new replication slot: {}", SLOT_NAME);
            } else {
                log.info("Using existing replication slot: {}", SLOT_NAME);
            }
            listenToReplicationSlot();
        } catch (Exception e) {
            log.error("初始化复制槽时发生错误", e);
        }
    }
    private boolean isSlotExists(String slotName) {
        String query = "SELECT COUNT(*) FROM pg_replication_slots WHERE slot_name = ?";
        Integer count = jdbcTemplate.queryForObject(query, Integer.class, slotName);
        return count != null && count > 0;
    }
    private void createReplicationSlot(String slotName) {
        String query = "SELECT pg_create_logical_replication_slot(?, 'test_decoding')";
        jdbcTemplate.update(query, slotName);
    }
    public void listenToReplicationSlot() {
        String query = "SELECT data FROM pg_logical_slot_get_changes(?, NULL, NULL)";
        int retryCount = 0;
        while (running) {
            try {
                List<String> changes = jdbcTemplate.queryForList(query, String.class, SLOT_NAME);
                for (String change : changes) {
                    try {
                        changeLogProcessor.processChangeLog(change);
                    } catch (Exception e) {
                        log.error("处理变更时发生错误: {}", change, e);
                    }
                }
                retryCount = 0; // 重置重试计数
                Thread.sleep(POLL_INTERVAL);
            } catch (InterruptedException e) {
                log.info("复制监听器被中断");
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("监听复制槽时发生错误", e);
                retryCount++;
                if (retryCount >= MAX_RETRIES) {
                    log.error("达到最大重试次数，停止监听");
                    break;
                }
                try {
                    Thread.sleep(POLL_INTERVAL * retryCount); // 指数退避
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
    public void stop() {
        running = false;
    }
}