| | |
| | |
|
| | |
|
| | | public void doStart() {
|
| | | try {
|
| | | Message message = null;
|
| | | try {
|
| | | message = connector.getWithoutAck(endpointInstance.getBatchSize()); // 获取指定数量的数据
|
| | |
| | | return;
|
| | | }
|
| | | // 如果重试次数小于设置的,则修改
|
| | | if (local_retry_count < endpointInstance.getRetryCount()) local_retry_count = endpointInstance.getRetryCount();
|
| | | 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());
|
| | |
| | | }
|
| | | }
|
| | | connector.ack(message.getId()); // 提交确认
|
| | | } catch (Exception exc) {
|
| | | logger.error("[DmlMessageTransponderContainer_doStart] Pull data message exception,errorMessage:{}", exc.getLocalizedMessage());
|
| | | // 防止删除消息时发生错误,或者拉取消息失败等情况
|
| | | exc.printStackTrace();
|
| | | }
|
| | | }
|
| | |
|
| | | public DmlMessageTransponderContainer(
|