切加颜分支

guang

2024-06-15 b9f78f6463160cb30f6cc276531c3ecc4c7834ee

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
package com.hz.canal.starter.factory;
 
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.exception.CanalClientException;
import com.hz.canal.starter.configuration.CanalAutoConfigurationProperties;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
 
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
 
 
public class CanalConnectorFactory {
 
    /**
     * 创建 CanalConnector
     *
     * @param destination
     * @param endpointInstance
     * @return CanalConnector
     * @author wuqiong 2022-04-23 20:36
     */
    public static synchronized CanalConnector createConnector(
            String destination, CanalAutoConfigurationProperties.EndpointInstance endpointInstance) {
        Assert.isTrue(StringUtils.hasText(destination), "destination is null , please check ");
        Assert.isTrue(endpointInstance != null, "endpoint instance is null , please check ");
        CanalConnector connector;
        if (endpointInstance.isClusterEnabled()) {
            if (!StringUtils.hasText(endpointInstance.getZookeeperAddress()))
                throw new CanalClientException("zookeeper address is null");
            List<SocketAddress> addresses = new ArrayList<>();
            for (String s : endpointInstance.getZookeeperAddress().split(",")) {
                String[] split = s.split(":");
                if (split.length != 2)
                    throw new CanalClientException("error parsing zookeeper address:" + s);
                addresses.add(new InetSocketAddress(split[0], Integer.parseInt(split[1])));
            }
            connector = CanalConnectors.newClusterConnector(
                    addresses, destination, endpointInstance.getUserName(), endpointInstance.getPassword());
        } else {
            connector = CanalConnectors.newSingleConnector(
                    new InetSocketAddress(endpointInstance.getHost(), endpointInstance.getPort()),
                    destination, endpointInstance.getUserName(), endpointInstance.getPassword());
        }
        return connector;
    }
}