duxinglangzi
2022-09-06 ddae546da9487622631d47133c962e7d870982de
增加错误消息日志打印
1个文件已修改
71 ■■■■ 已修改文件
src/main/java/com/duxinglangzi/canal/starter/container/DmlMessageTransponderContainer.java 71 ●●●● 补丁 | 查看 | 原始文档 | blame | 历史
src/main/java/com/duxinglangzi/canal/starter/container/DmlMessageTransponderContainer.java
@@ -51,42 +51,49 @@
    public void doStart() {
        Message message = null;
        try {
            message = connector.getWithoutAck(endpointInstance.getBatchSize()); // 获取指定数量的数据
        } catch (Exception clientException) {
            logger.error("[DmlMessageTransponderContainer] error msg : ", clientException);
            if (local_retry_count > 0) {
                // 重试次数减一
                local_retry_count = local_retry_count - 1;
                sleep(endpointInstance.getAcquireInterval());
            } else {
                // 重试次数 <= 0 时,直接终止线程
                logger.error("[DmlMessageTransponderContainer] retry count is zero ,  " +
                                "thread interrupt , current connector host: {} , port: {} ",
                        endpointInstance.getHost(), endpointInstance.getPort());
                Thread.currentThread().interrupt();
            }
            return;
        }
        // 如果重试次数小于设置的,则修改
        if (local_retry_count < endpointInstance.getRetryCount()) local_retry_count = endpointInstance.getRetryCount();
        List<CanalEntry.Entry> entries = message.getEntries();
        if (message.getId() == -1 || entries.isEmpty()) {
            sleep(endpointInstance.getAcquireInterval());
            return;
        }
        for (CanalEntry.Entry entry : entries) {
            Message message = null;
            try {
                consumer(entry);
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("[DmlMessageTransponderContainer_doStart] CanalEntry.Entry consumer error ", e);
                // connector.rollback(message.getId()); // 目前先不处理失败, 无需回滚数据
                // return;
                message = connector.getWithoutAck(endpointInstance.getBatchSize()); // 获取指定数量的数据
            } catch (Exception clientException) {
                logger.error("[DmlMessageTransponderContainer] error msg : ", clientException);
                if (local_retry_count > 0) {
                    // 重试次数减一
                    local_retry_count = local_retry_count - 1;
                    sleep(endpointInstance.getAcquireInterval());
                } else {
                    // 重试次数 <= 0 时,直接终止线程
                    logger.error("[DmlMessageTransponderContainer] retry count is zero ,  " +
                                    "thread interrupt , current connector host: {} , port: {} ",
                            endpointInstance.getHost(), endpointInstance.getPort());
                    Thread.currentThread().interrupt();
                }
                return;
            }
            // 如果重试次数小于设置的,则修改
            if (local_retry_count < endpointInstance.getRetryCount())
                local_retry_count = endpointInstance.getRetryCount();
            List<CanalEntry.Entry> entries = message.getEntries();
            if (message.getId() == -1 || entries.isEmpty()) {
                sleep(endpointInstance.getAcquireInterval());
                return;
            }
            for (CanalEntry.Entry entry : entries) {
                try {
                    consumer(entry);
                } catch (Exception e) {
                    e.printStackTrace();
                    logger.error("[DmlMessageTransponderContainer_doStart] CanalEntry.Entry consumer error ", e);
                    // connector.rollback(message.getId()); // 目前先不处理失败, 无需回滚数据
                    // return;
                }
            }
            connector.ack(message.getId()); // 提交确认
        } catch (Exception exc) {
            logger.error("[DmlMessageTransponderContainer_doStart] Pull data message exception,errorMessage:{}", exc.getLocalizedMessage());
            // 防止删除消息时发生错误,或者拉取消息失败等情况
            exc.printStackTrace();
        }
        connector.ack(message.getId()); // 提交确认
    }
    public DmlMessageTransponderContainer(