canal1.8报错

Question

Exception in thread "main" com.alibaba.otter.canal.protocol.exception.CanalClientException: failed to subscribe with reason: something goes wrong with channel:[id: 0x11ba2eaa, /192.168.0.38:7546 => /192.168.0.38:11111], exception=com.alibaba.otter.canal.server.exception.CanalServerException: destination:example should start first
学习canal的时候java集成一直报错,不知道为什么,帮忙看看
其中canal.serverMode = tcp,canal.destinations = example都是走的默认的,
example配置如图

查看了canal.log日志有报错情况

Answer 1

该回答引用自GPT-3.5, 由博主 GISer Liu 编写：

问题分析

用户在使用Canal进行Java集成时遇到了报错问题。报错信息显示：Exception in thread "main" com.alibaba.otter.canal.protocol.exception.CanalClientException: failed to subscribe with reason: something goes wrong with channel:[id: 0x11ba2eaa, /192.168.0.38:7546 => /192.168.0.38:11111], exception=com.alibaba.otter.canal.server.exception.CanalServerException: destination:example should start first。从报错信息来看，问题主要集中在Canal客户端订阅时，Canal服务器端出现异常，提示destination:example should start first。

解决方案

1. 确认Canal服务器配置

首先，需要确认Canal服务器的配置是否正确。特别是canal.destinations和canal.serverMode这两个配置项。

• canal.destinations：指定Canal实例的目标，这里应该是example。
• canal.serverMode：指定Canal服务器的模式，这里应该是tcp。

请检查Canal服务器的配置文件canal.properties，确保以下配置正确：

canal.destinations = example
canal.serverMode = tcp

2. 启动Canal实例

报错信息提示destination:example should start first，这意味着Canal实例example没有启动。请确保Canal实例example已经启动。

启动Canal实例的命令如下：

sh bin/startup.sh -d example

3. 检查Canal实例配置

确认Canal实例example的配置文件example/instance.properties是否正确。特别是以下几个关键配置项：

• canal.instance.mysql.slaveId：MySQL的slave ID，需要唯一。
• canal.instance.master.address：MySQL主库的地址。
• canal.instance.dbUsername：MySQL的用户名。
• canal.instance.dbPassword：MySQL的密码。

确保这些配置项正确无误。

4. 检查Canal客户端代码

确认Canal客户端代码是否正确。以下是一个简单的Canal客户端示例代码：

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;

public class CanalClientExample {
    public static void main(String[] args) {
        // 创建Canal连接器
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("192.168.0.38", 11111),
                "example",
                "",
                ""
        );

        try {
            // 连接Canal服务器
            connector.connect();
            // 订阅指定的destination
            connector.subscribe(".*\\..*");
            // 循环处理消息
            while (true) {
                Message message = connector.getWithoutAck(100); // 获取指定数量的消息
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    // 没有消息，等待一会儿
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    // 处理消息
                    printEntry(message.getEntries());
                }
                // 确认消息已处理
                connector.ack(batchId);
            }
        } finally {
            // 关闭连接
            connector.disconnect();
        }
    }

    private static void printEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChage = null;
                try {
                    rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
                }

                CanalEntry.EventType eventType = rowChage.getEventType();
                System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                        eventType));

                for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                    if (eventType == CanalEntry.EventType.DELETE) {
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == CanalEntry.EventType.INSERT) {
                        printColumn(rowData.getAfterColumnsList());
                    } else {
                        System.out.println("-------> before");
                        printColumn(rowData.getBeforeColumnsList());
                        System.out.println("-------> after");
                        printColumn(rowData.getAfterColumnsList());
                    }
                }
            }
        }
    }

    private static void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}

5. 检查网络连接

确保Canal客户端能够正常连接到Canal服务器。可以通过telnet命令检查端口是否开放：

telnet 192.168.0.38 11111

如果连接失败，请检查防火墙设置，确保Canal服务器的端口是开放的。

总结

通过以上步骤，可以逐步排查并解决Canal客户端报错问题。首先确认Canal服务器的配置和实例是否正确启动，然后检查Canal客户端代码是否正确，最后确保网络连接正常。希望这些步骤能帮助您解决问题。

如果该回答解决了您的问题，请采纳！如果没有，请私信联系或评论您的疑惑