springboot整合easy-es和canbal、kafka实现mysql数据同步

Spring Boot可以很方便地整合各种组件和框架，包括Elasticsearch、Canal和Kafka。下面简单介绍一下如何使用Spring Boot整合这三个组件实现MySQL数据同步到Elasticsearch的功能。

1. 集成Easy Elasticsearch

首先需要在pom.xml中引入Easy Elasticsearch的依赖：

<dependency>
    <groupId>com.jdon</groupId>
    <artifactId>easy-elasticsearch</artifactId>
    <version>1.0.0</version>
</dependency>

然后在application.properties中配置Elasticsearch的地址：

spring.elasticsearch.rest.uris=http://localhost:9200

2. 集成Canal

Canal是阿里巴巴开源的一款MySQL数据增量订阅&消费组件，可以实时监听MySQL的binlog并将数据同步到其他存储介质，比如Kafka或Elasticsearch。

在pom.xml中引入Canal的依赖：

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

然后在application.properties中配置Canal的参数：

canal.server.host=localhost
canal.server.port=11111
canal.destination=test
canal.username=
canal.password=

3. 集成Kafka

Kafka是一款分布式的消息队列，可以将数据异步地发送到其他系统或存储介质。

在pom.xml中引入Kafka的依赖：

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.3.4.RELEASE</version>
</dependency>

然后在application.properties中配置Kafka的参数：

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=test-group

4. 实现数据同步

首先需要创建一个Canal客户端，实现Canal的监听器接口，监听到MySQL的binlog变化时将数据发送到Kafka。

@Component
public class CanalClient implements CanalEventListener {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void onEvent(CanalEvent canalEvent) {
        List<CanalEntry.Entry> entries = canalEvent.getEntries();
        for (CanalEntry.Entry entry : entries) {
            CanalEntry.EntryType entryType = entry.getEntryType();
            if (entryType == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChange;
                try {
                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
                }
                if (rowChange != null) {
                    String tableName = entry.getHeader().getTableName();
                    for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                        Map<String, String> dataMap = new HashMap<>();
                        for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
                            dataMap.put(column.getName(), column.getValue());
                        }
                        kafkaTemplate.send(tableName, new Gson().toJson(dataMap));
                    }
                }
            }
        }
    }
}

然后创建一个Kafka消费者，将数据从Kafka读取出来，再通过Easy Elasticsearch将数据同步到Elasticsearch。

@Component
public class KafkaConsumer {

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;

    @KafkaListener(topics = "test")
    public void processMessage(String message) {
        Gson gson = new Gson();
        Type type = new TypeToken<Map<String, String>>(){}.getType();
        Map<String, String> dataMap = gson.fromJson(message, type);
        IndexQuery indexQuery = new IndexQueryBuilder()
                .withId(dataMap.get("id"))
                .withObject(dataMap)
                .build();
        elasticsearchTemplate.index(indexQuery);
    }
}

最后启动Spring Boot应用程序，就能实现MySQL数据同步到Elasticsearch的功能了。