| | |
|---|
| | |
|
|---|
| | |
|
|---|
| | | 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(
|
|---|
