# application.properties

# 數(shù)據(jù)源1：官網(wǎng)
spring.datasource.primary.url=jdbc:mysql://localhost:3306/website_db
spring.datasource.primary.username=root
spring.datasource.primary.password=password

# 數(shù)據(jù)源2：移動(dòng)App
spring.datasource.secondary.url=jdbc:mysql://localhost:3306/app_db
spring.datasource.secondary.username=root
spring.datasource.secondary.password=password

# 數(shù)據(jù)源3：第三方平臺(tái)
spring.datasource.tertiary.url=jdbc:mysql://localhost:3306/thirdparty_db
spring.datasource.tertiary.username=root
spring.datasource.tertiary.password=password

<!-- pom.xml -->
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-embedded</artifactId>
    <version>1.8.0.Final</version>
</dependency>

import io.debezium.config.Configuration;
import io.debezium.embedded.EmbeddedEngine;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import javax.sql.DataSource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@SpringBootApplication
public class TicketSyncApplication implements CommandLineRunner {

    @Autowired
    private DataSource primaryDataSource; // 官網(wǎng)數(shù)據(jù)源

    @Autowired
    private DataSource secondaryDataSource; // 移動(dòng)App數(shù)據(jù)源

    @Autowired
    private DataSource tertiaryDataSource; // 第三方平臺(tái)數(shù)據(jù)源
  
  	private final JdbcTemplate primaryJdbcTemplate;
    private final JdbcTemplate secondaryJdbcTemplate;
    private final JdbcTemplate tertiaryJdbcTemplate;

    public static void main(String[] args) {
        SpringApplication.run(TicketSyncApplication.class, args);
    }

  @Autowired
    public TicketSyncApplication(DataSource primaryDataSource, DataSource secondaryDataSource, DataSource tertiaryDataSource) {
        
  	 		this.primaryJdbcTemplate = new JdbcTemplate(primaryDataSource);
        this.secondaryJdbcTemplate = new JdbcTemplate(secondaryDataSource);
        this.tertiaryJdbcTemplate = new JdbcTemplate(tertiaryDataSource);
    }
  
    @Override
    public void run(String... args) throws Exception {
        Configuration config = Configuration.create()
                .with("name", "ticket-sync-connector")
                .with("connector.class", "io.debezium.connector.mysql.MySqlConnector")
                .with("database.hostname", "localhost")
                .with("database.port", "3306")
                .with("database.user", "root")
                .with("database.password", "password")
                .with("database.server.id", "85744")
                .with("database.server.name", "ticket_server")
                .with("database.whitelist", "website_db")
                .with("table.whitelist", "website_db.tickets")
                .with("database.history", "io.debezium.relational.history.FileDatabaseHistory")
                .with("database.history.file.filename", "/tmp/dbhistory.dat")
                .build();

        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
                .using(config.asProperties())
                .notifying(this::handleEvent)
                .build();

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.execute(engine);
    }

    private void handleEvent(ChangeEvent<String, String> event) {
        // 處理數(shù)據(jù)變化事件
        String key = event.key();
        String value = event.value();
        System.out.println("Change detected: " + key + " = " + value);

        // 解析變化數(shù)據(jù)并同步到各個(gè)渠道
        syncDataToChannels(key, value);
    }

    @Transactional
    public void syncDataToChannels(String key, String value) {
        // 解析變化的數(shù)據(jù)
        // 假設(shè)key是主鍵，value是JSON格式的票務(wù)信息
        Map<String, Object> ticketData = parseValue(value);

        // 從解析的數(shù)據(jù)中獲取必要字段
        String ticketId = (String) ticketData.get("ticketId");
        String ticketInfo = (String) ticketData.get("ticketInfo");

        // 同步到移動(dòng)App數(shù)據(jù)源
        updateTicketInDataSource(secondaryJdbcTemplate, ticketId, ticketInfo);

        // 同步到第三方平臺(tái)數(shù)據(jù)源
        updateTicketInDataSource(tertiaryJdbcTemplate, ticketId, ticketInfo);
    }

    private Map<String, Object> parseValue(String value) {
        // 解析JSON字符串為Map
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            return objectMapper.readValue(value, new TypeReference<Map<String, Object>>() {});
        } catch (IOException e) {
            throw new RuntimeException("Failed to parse value: " + value, e);
        }
    }

    private void updateTicketInDataSource(JdbcTemplate jdbcTemplate, String ticketId, String ticketInfo) 		{
        String updateQuery = "UPDATE tickets SET ticket_info = ? WHERE ticket_id = ?";
        jdbcTemplate.update(updateQuery, ticketInfo, ticketId);
    }
}