利用Binlog和Kafka实时同步mysql数据到Elasticsearch(四) - 消费Kafka消息同步数据到ES

目录

1、利用Binlog和Kafka实时同步mysql数据到Elasticsearch(一) - 开启Binlog日志
2、利用Binlog和Kafka实时同步mysql数据到Elasticsearch(二) - 安装并运行Kafka
3、利用Binlog和Kafka实时同步mysql数据到Elasticsearch(三) - Binlog日志生产消息到Kafka
4、利用Binlog和Kafka实时同步mysql数据到Elasticsearch(四) - 消费Kafka消息同步数据到ES


前言

- 项目模块

BinlogMiddleware

1、binlog中间件,负责解析binlog,把变动的数据以json形式发送到kafka队列。

KafkaMiddleware

2、kafka中间件,负责消费kafka队列中的Message,把数据写入Elasticsearch中。

- 基础服务

(1)Mysql
(2)Kafka(用于存放mysql变动消息,存放于Kafka队列)
(3)Elasticsearch

- 项目源码

码云:https://gitee.com/OrgXxxx/SyncMysqlToElasticsearch

简介:

KafkaMiddleware服务主要负责消费Kafka队列消息,并将其同步到Elastcsearch(及Kafka消费者)。

  • 本示例模拟监听teemoliu数据库的user、role表。为了方便表结构设计的很简单,均只含有id、name两个属性。
  • 中间件读取Kafka队列的消息格式如下:
{"event":"teemoliu.user.update","value":[1,"TeemoLiu"]}
{"event":"teemoliu.role.insert","value":[1,"管理员"]}
  • 项目结构如下:


    image.png

1、导入maven引用

        <!--kafka-->
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <version>2.2.4.RELEASE</version>
        </dependency>
        <!-- elasticsearch http api client -->
        <dependency>
            <groupId>io.searchbox</groupId>
            <artifactId>jest</artifactId>
            <version>5.3.3</version>
        </dependency>
        <!--fastjson-->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.49</version>
        </dependency>

2、配置文件如下:

#停用服务端口
spring.main.web-environment=false
#============== kafka ===================
# 指定kafka 代理地址,可以多个
spring.kafka.bootstrap-servers=localhost:9092
#=============== consumer  =======================
# 指定默认消费者group id
spring.kafka.consumer.group-id=consumer1
spring.kafka.consumer.auto-offset-reset=latest
#spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=true
#spring.kafka.consumer.auto-commit-interval=100
# 指定消息key和消息体的编解码方式
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
# 消息JSON格式化模板
es.data.format.user={"id":"0","name":"1"}
es.data.format.role={"id":"0","name":"1"}

3、初始化Jest客户端

public class EsJestClient {

    private static JestClient client;

    /**
     * 获取客户端
     *
     * @return jestclient
     */
    public static synchronized JestClient getClient() {
        if (client == null) {
            build();
        }
        return client;
    }

    /**
     * 关闭客户端
     */
    public static void close(JestClient client) {
        if (!Objects.isNull(client)) {
            try {
                client.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 建立连接
     */
    private static void build() {
        JestClientFactory factory = new JestClientFactory();
        factory.setHttpClientConfig(
                new HttpClientConfig
                        .Builder(Config.ES_HOST)
                        .multiThreaded(true)
                        //一个route 默认不超过2个连接  路由是指连接到某个远程注解的个数。总连接数=route个数 * defaultMaxTotalConnectionPerRoute
                        .defaultMaxTotalConnectionPerRoute(2)
                        //所有route连接总数
                        .maxTotalConnection(2)
                        .connTimeout(10000)
                        .readTimeout(10000)
                        .gson(new GsonBuilder()
                                .setDateFormat("yyyy-MM-dd HH:mm:ss")
                                .create())
                        .build()
        );
        client = factory.getObject();
    }

}

4、实现Kafka批量消费

@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Value("${spring.kafka.consumer.group-id}")
    String groupId;
    @Value("${spring.kafka.bootstrap-servers}")
    String bootstrapServers;
    @Value("${spring.kafka.consumer.auto-offset-reset}")
    String autoOffsetReset;
    
    @Bean
    KafkaListenerContainerFactory<?> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new
                ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs()));
        factory.setBatchListener(true);
        return factory;
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);//每一批数量
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 120000);
        props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, 180000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return props;
    }

}

5、实现ES 通用业务逻辑层

@Service
public class ESService{

    private JestClient client;

    public ESService(JestClient client) {
        this.client = client;
    }

    public boolean update(String id, String esType, Object object) {
        Index index = new Index.Builder(object).index(Config.ES_INDICES).type(esType).id(id).refresh(true).build();
        try {
            JestResult result = client.execute(index);
            return result != null && result.isSucceeded();
        } catch (Exception ignore) {
        }
        return false;
    }

    public Index getUpdateIndex(String id, String esType, Object object) {
        return new Index.Builder(object).index(Config.ES_INDICES).type(esType).id(id).refresh(true).build();
    }

    public Delete getDeleteIndex(String id, String esType) {
        return new Delete.Builder(id).index(Config.ES_INDICES).type(esType).build();
    }

    public boolean executeESClientRequest(List indexList, String esType) {
        Bulk bulk = new Bulk.Builder()
                .defaultIndex(Config.ES_INDICES)
                .defaultType(esType)
                .addAction(indexList)
                .build();
        indexList.clear();
        try {
            JestResult result = client.execute(bulk);
            return result != null && result.isSucceeded();
        } catch (Exception ignore) {
        }
        return false;
    }

    public boolean delete(String id, String esType) {
        try {
            DocumentResult result = client.execute(new Delete.Builder(id)
                    .index(Config.ES_INDICES)
                    .type(esType)
                    .build());
            return result.isSucceeded();
        } catch (Exception e) {
            throw new RuntimeException("delete exception", e);
        }
    }
}

6、实现消费逻辑

@Component
public class JsonConsumer {

    @Value("${es.data.format.user}")
    String userFormat;
    @Value("${es.data.format.role}")
    String roleFormat;

    JestClient client = EsJestClient.getClient();
    ESService documentDao = new ESService(client);

    @KafkaListener(topics = Config.KAFKA_JSON_TOPICS, id = Config.KAFKA_JSON_ID, containerFactory = "batchFactory")
    public void listen(List<ConsumerRecord<?, ?>> list) {
        List<String> messages = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : list) {
            Optional<?> kafkaMessage = Optional.ofNullable(record.value());
            // 获取消息
            kafkaMessage.ifPresent(o -> messages.add(o.toString()));
        }
        if (messages.size() > 0) {
            // 更新索引
            updateES(messages);
        }
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取ES的TYPE
     *
     * @param tableName
     * @return
     */
    private String getESType(String tableName) {
        String esType = "";
        switch (tableName) {
            case "role": {
                esType = Config.ES_ROLE_TYPE;
                break;
            }
            case "user": {
                esType = Config.ES_USER_TYPE;
                break;
            }
        }
        return esType;
    }

    /**
     * 获取消息JSON解析格式
     *
     * @param tableName
     * @return
     */
    private String getJsonFormat(String tableName) {
        String format = "";
        switch (tableName) {
            case "role": {
                format = roleFormat;
                break;
            }
            case "user": {
                format = userFormat;
                break;
            }
        }
        return format;
    }


    /**
     * 获取解析后的ES对象
     *
     * @param message
     * @param tableName
     * @return
     */
    private JSONObject getESObject(JSONArray message, String tableName) {
        JSONObject resultObject = new JSONObject();
        String format = getJsonFormat(tableName);
        if (!format.isEmpty()) {
            JSONObject jsonFormatObject = JSON.parseObject(format);
            for (String key : jsonFormatObject.keySet()) {
                String[] formatValues = jsonFormatObject.getString(key).split(",");
                if (formatValues.length < 2) {
                    resultObject.put(key, message.get(jsonFormatObject.getInteger(key)));
                } else {
                    Object object = message.get(Integer.parseInt(formatValues[0]));
                    if (object == null) {
                        String[] array = {};
                        resultObject.put(key, array);
                    } else {
                        String objectStr = message.get(Integer.parseInt(formatValues[0])).toString();
                        String[] result = objectStr.split(formatValues[1]);
                        resultObject.put(key, result);
                    }
                }
            }
        }
        return resultObject;
    }


    /**
     * 更新ES索引
     *
     * @param messages
     */
    private void updateES(List<String> messages) {
        List<Index> updateUserList = new ArrayList<>();
        List<Index> updateRoleList = new ArrayList<>();
        List<Delete> deleteUserList = new ArrayList<>();
        List<Delete> deleteRoleList = new ArrayList<>();
        for (String message : messages) {
            JSONObject result = null;
            try {
                result = JSON.parseObject(message);
            } catch (Exception e) {
                continue;
            }
            // 获取事件类型 event:"wtv3.videos.insert"
            String event = (String) result.get("event");
            String[] eventArray = event.split("\\.");
            String tableName = eventArray[1];
            String eventType = eventArray[2];
            // 获取具体数据
            JSONArray valueStr = (JSONArray) result.get("value");
            // 转化为对应格式的json字符串
            JSONObject object = getESObject(valueStr, tableName);
            // 获取ES的type
            String esType = getESType(tableName);
            switch (eventType) {
                case "insert": {
                    appendUpdateList(updateUserList, updateRoleList, object, esType);
                    break;
                }
                case "update": {
                    // 更新videos
                    appendUpdateList(updateUserList, updateRoleList, object, esType);
                    break;
                }
                case "delete": {
                    // 删除videos
                    appendDeleteList(deleteUserList, deleteRoleList, object, esType);
                    break;
                }
            }
        }
        if (updateUserList.size() > 0) {
            documentDao.executeESClientRequest(updateUserList, Config.ES_USER_TYPE);
        }
        if (updateRoleList.size() > 0) {
            documentDao.executeESClientRequest(updateRoleList, Config.ES_ROLE_TYPE);
        }
        if (deleteUserList.size() > 0) {
            documentDao.executeESClientRequest(deleteUserList, Config.ES_USER_TYPE);
        }
        if (deleteRoleList.size() > 0) {
            documentDao.executeESClientRequest(deleteRoleList, Config.ES_ROLE_TYPE);
        }
    }

    private void appendDeleteList(List<Delete> userList, List<Delete> roleList, JSONObject object, String esType) {
        switch (esType) {
            case Config.ES_USER_TYPE: {
                userList.add(documentDao.getDeleteIndex(object.get("id").toString(), esType));
                break;
            }
            case Config.ES_ROLE_TYPE: {
                roleList.add(documentDao.getDeleteIndex(object.get("id").toString(), esType));
                break;
            }
        }
    }

    private void appendUpdateList(List<Index> userList, List<Index> roleList, JSONObject object, String esType) {
        switch (esType) {
            case Config.ES_USER_TYPE: {
                userList.add(documentDao.getUpdateIndex(object.get("id").toString(), esType, object));
                break;
            }
            case Config.ES_ROLE_TYPE: {
                roleList.add(documentDao.getUpdateIndex(object.get("id").toString(), esType, object));
                break;
            }
        }
    }

}

7、运行结果(源码在前言)

Mysql-role表数据
Mysql-user表数据
Elasticsearch同步数据
最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,324评论 5 476
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,303评论 2 381
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,192评论 0 337
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,555评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,569评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,566评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,927评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,583评论 0 257
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,827评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,590评论 2 320
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,669评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,365评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,941评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,928评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,159评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,880评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,399评论 2 342