实时同步mysql数据到Elasticsearch（3.消费Ka

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、运行结果（源码在前言）