修改了listener方法的参数

duxinglangzi

2022-08-15 c83c1b4ee7ff9c01a7a67855863c281589f39c72

 src/main/java/com/duxinglangzi/canal/starter/container/DmlMessageTransponderContainer.java

@@ -5,6 +5,7 @@
import com.alibaba.otter.canal.protocol.Message;

import com.duxinglangzi.canal.starter.configuration.CanalAutoConfigurationProperties;

import com.duxinglangzi.canal.starter.configuration.CanalListenerEndpointRegistrar;

import com.duxinglangzi.canal.starter.mode.CanalMessage;

import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



@@ -23,16 +24,19 @@
    private CanalConnector connector;

    private CanalAutoConfigurationProperties.EndpointInstance endpointInstance;

    private List<CanalListenerEndpointRegistrar> registrars = new ArrayList<>();

    private Set<CanalEntry.EventType> SUPPORT_ALL_TYPES = new HashSet<>();

    private Set<CanalEntry.EventType> support_all_types = new HashSet<>();

    private Integer local_retry_count;





    public void initConnect() {

        try {

            // init supportAllTypes

            registrars.forEach(e -> SUPPORT_ALL_TYPES.addAll(Arrays.asList(e.getEventType())));

            registrars.forEach(e -> support_all_types.addAll(Arrays.asList(e.getEventType())));

            connector.connect();

            connector.subscribe(endpointInstance.getSubscribe());

            connector.rollback();

            // 初始化本地重试次数

            local_retry_count = endpointInstance.getRetryCount();

        } catch (Exception e) {

            logger.error("[DmlMessageTransponderContainer_initConnect] canal client connect error .", e);

            setRunning(false);

@@ -52,18 +56,21 @@
            message = connector.getWithoutAck(endpointInstance.getBatchSize()); // 获取指定数量的数据

        } catch (Exception clientException) {

            logger.error("[DmlMessageTransponderContainer] error msg : ", clientException);

            endpointInstance.setRetryCount(endpointInstance.getRetryCount() - 1);

            if (endpointInstance.getRetryCount() < 0) {

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

                disconnect();

            } else {

                sleep(endpointInstance.getAcquireInterval());

            }

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

@@ -103,7 +110,7 @@
        // 忽略 ddl 语句

        if (rowChange.hasIsDdl() && rowChange.getIsDdl()) return;

        CanalEntry.EventType eventType = rowChange.getEventType();

        if (!SUPPORT_ALL_TYPES.contains(eventType)) return;

        if (!support_all_types.contains(eventType)) return;

        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {

            registrars

                    .stream()

@@ -113,7 +120,7 @@
                            eventType))

                    .forEach(element -> {

                        try {

                            element.getMethod().invoke(element.getBean(), eventType, rowData);

                            element.getMethod().invoke(element.getBean(), new CanalMessage(entry.getHeader(), eventType, rowData));

                        } catch (IllegalAccessException | InvocationTargetException e) {

                            logger.error("[DmlMessageTransponderContainer_consumer] RowData Callback Method invoke error message", e);

                            throw new RuntimeException("RowData Callback Method invoke error message： " + e.getMessage(), e);