From 79552e0a37980824ca77f6eb49ea17db07d0fd87 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=80=90=E5=B0=8F=E5=BF=83?= Date: Mon, 16 Sep 2019 12:43:02 +0800 Subject: [PATCH] =?UTF-8?q?rabbitmq(amqp)MQ=E6=94=AF=E6=8C=81=20(#2156)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../adapter/support/CanalClientConfig.java | 41 ++- client-adapter/launcher/pom.xml | 10 + .../launcher/loader/CanalAdapterLoader.java | 23 ++ .../loader/CanalAdapterRabbitMQWorker.java | 92 ++++++ .../src/main/resources/application.yml | 3 + client/pom.xml | 8 + .../otter/canal/client/CanalConnector.java | 1 + .../{rocketmq => }/ConsumerBatchMessage.java | 2 +- .../rabbitmq/AliyunCredentialsProvider.java | 54 ++++ .../rabbitmq/RabbitMQCanalConnector.java | 299 ++++++++++++++++++ .../rocketmq/RocketMQCanalConnector.java | 1 + .../otter/canal/deployer/CanalConstants.java | 5 + .../otter/canal/deployer/CanalStarter.java | 28 ++ deployer/src/main/resources/canal.properties | 5 + deployer/src/main/resources/logback.xml | 5 +- .../canal/example/SimpleCanalClientTest.java | 10 +- pom.xml | 11 + server/pom.xml | 8 + .../otter/canal/common/MQProperties.java | 45 +++ .../rabbitmq/AliyunCredentialsProvider.java | 55 ++++ .../canal/rabbitmq/CanalRabbitMQProducer.java | 119 +++++++ 21 files changed, 817 insertions(+), 8 deletions(-) create mode 100644 client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRabbitMQWorker.java rename client/src/main/java/com/alibaba/otter/canal/client/{rocketmq => }/ConsumerBatchMessage.java (95%) create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/AliyunCredentialsProvider.java create mode 100644 client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/RabbitMQCanalConnector.java create mode 100644 server/src/main/java/com/alibaba/otter/canal/rabbitmq/AliyunCredentialsProvider.java create mode 100644 server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java index 00939c58..466ed20f 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/CanalClientConfig.java @@ -33,6 +33,13 @@ public class CanalClientConfig { // aliyun ak/sk private String accessKey; private String secretKey; + // rabbitmq 账号密码 + private String username; + private String password; + // rabbitmq vhost + private String vhost = "/"; + + private Long resourceOwnerId; // 是否启用消息轨迹 private boolean enableMessageTrace; // 在使用阿里云商业化mq服务时，如果想使用云上消息轨迹功能，请设置此配置为true @@ -132,6 +139,38 @@ public class CanalClientConfig { this.secretKey = secretKey; } + public String getUsername() { + return username; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public String getVhost() { + return vhost; + } + + public void setVhost(String vhost) { + this.vhost = vhost; + } + + public Long getResourceOwnerId() { + return resourceOwnerId; + } + + public void setResourceOwnerId(Long resourceOwnerId) { + this.resourceOwnerId = resourceOwnerId; + } + public List getCanalAdapters() { return canalAdapters; } @@ -176,7 +215,7 @@ public class CanalClientConfig { private String instance; // 实例名 - private List groups; // 适配器分组列表 + private List groups; // 适配器分组列表 public String getInstance() { return instance; diff --git a/client-adapter/launcher/pom.xml b/client-adapter/launcher/pom.xml index 1b2163de..879a0cbc 100644 --- a/client-adapter/launcher/pom.xml +++ b/client-adapter/launcher/pom.xml @@ -71,6 +71,16 @@ 1.1.1 + + com.rabbitmq + amqp-client + 5.5.0 + + + com.alibaba.mq-amqp + mq-amqp-client + 1.0.3 + mysql diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java index 7623c048..51e0c3dd 100644 --- a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterLoader.java @@ -142,6 +142,29 @@ public class CanalAdapterLoader { canalAdapter.getInstance() + "-" + group.getGroupId()); } } + } else if ("rabbitMQ".equalsIgnoreCase(canalClientConfig.getMode())) { + // 初始化canal-client-rabbitMQ的适配器 + for (CanalClientConfig.CanalAdapter canalAdapter : canalClientConfig.getCanalAdapters()) { + for (CanalClientConfig.Group group : canalAdapter.getGroups()) { + List> canalOuterAdapterGroups = new ArrayList<>(); + List canalOuterAdapters = new ArrayList<>(); + for (OuterAdapterConfig config : group.getOuterAdapters()) { + loadAdapter(config, canalOuterAdapters); + } + canalOuterAdapterGroups.add(canalOuterAdapters); + CanalAdapterRabbitMQWorker rabbitMQWork = new CanalAdapterRabbitMQWorker(canalClientConfig, + canalOuterAdapterGroups, + canalAdapter.getInstance(), + group.getGroupId(), + canalClientConfig.getFlatMessage()); + canalMQWorker.put(canalAdapter.getInstance() + "-rabbitmq-" + group.getGroupId(), rabbitMQWork); + rabbitMQWork.start(); + + logger.info("Start adapter for canal-client mq topic: {} succeed", + canalAdapter.getInstance() + "-" + group.getGroupId()); + } + } + // CanalAdapterRabbitMQWork } } diff --git a/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRabbitMQWorker.java b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRabbitMQWorker.java new file mode 100644 index 00000000..0de7c468 --- /dev/null +++ b/client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/loader/CanalAdapterRabbitMQWorker.java @@ -0,0 +1,92 @@ +package com.alibaba.otter.canal.adapter.launcher.loader; + +import com.alibaba.otter.canal.client.adapter.OuterAdapter; +import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig; +import com.alibaba.otter.canal.client.adapter.support.Util; +import com.alibaba.otter.canal.client.rabbitmq.RabbitMQCanalConnector; +import org.apache.kafka.common.errors.WakeupException; + +import java.util.List; +import java.util.concurrent.ExecutorService; + +public class CanalAdapterRabbitMQWorker extends AbstractCanalAdapterWorker { + + private RabbitMQCanalConnector connector; + private String topic; + private boolean flatMessage; + + public CanalAdapterRabbitMQWorker(CanalClientConfig canalClientConfig, List> canalOuterAdapters, + String topic, String groupId, boolean flatMessage){ + super(canalOuterAdapters); + this.canalClientConfig = canalClientConfig; + this.topic = topic; + this.flatMessage = flatMessage; + this.canalDestination = topic; + this.groupId = groupId; + this.connector = new RabbitMQCanalConnector(canalClientConfig.getMqServers(), + canalClientConfig.getVhost(), + topic, + canalClientConfig.getAccessKey(), + canalClientConfig.getSecretKey(), + canalClientConfig.getUsername(), + canalClientConfig.getPassword(), + canalClientConfig.getResourceOwnerId(), + flatMessage); + } + + @Override + protected void process() { + while (!running) { + try { + Thread.sleep(1000); + } catch (InterruptedException ignored) { + } + } + + ExecutorService workerExecutor = Util.newSingleThreadExecutor(5000L); + int retry = canalClientConfig.getRetries() == null + || canalClientConfig.getRetries() == 0 ? 1 : canalClientConfig.getRetries(); + long timeout = canalClientConfig.getTimeout() == null ? 30000 : canalClientConfig.getTimeout(); // 默认超时30秒 + while (running) { + try { + syncSwitch.get(canalDestination); + logger.info("=============> Start to connect topic: {} <=============", this.topic); + connector.connect(); + logger.info("=============> Start to subscribe topic: {}<=============", this.topic); + connector.subscribe(); + logger.info("=============> Subscribe topic: {} succeed<=============", this.topic); + while (running) { + boolean status = syncSwitch.status(canalDestination); + if (!status) { + connector.disconnect(); + break; + } + if (retry == -1) { + retry = Integer.MAX_VALUE; + } + for (int i = 0; i < retry; i++) { + if (!running) { + break; + } + if (mqWriteOutData(retry, timeout, i, flatMessage, connector, workerExecutor)) { + break; + } + } + } + } catch (Exception e) { + logger.error(e.getMessage(), e); + } + } + + workerExecutor.shutdown(); + + try { + connector.unsubscribe(); + } catch (WakeupException e) { + // No-op. Continue process + } + connector.disconnect(); + logger.info("=============> Disconnect topic: {} <=============", this.topic); + + } +} diff --git a/client-adapter/launcher/src/main/resources/application.yml b/client-adapter/launcher/src/main/resources/application.yml index 38020409..f0f89a5e 100644 --- a/client-adapter/launcher/src/main/resources/application.yml +++ b/client-adapter/launcher/src/main/resources/application.yml @@ -18,6 +18,9 @@ canal.conf: timeout: accessKey: secretKey: + username: + password: + vhost: # srcDataSources: # defaultDS: # url: jdbc:mysql://127.0.0.1:3306/mytest?useUnicode=true diff --git a/client/pom.xml b/client/pom.xml index ad2bce68..0ccf06eb 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -109,6 +109,14 @@ org.apache.rocketmq rocketmq-acl + + com.rabbitmq + amqp-client + + + com.alibaba.mq-amqp + mq-amqp-client + org.apache.kafka diff --git a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java index 7f3eeb80..4211b9e5 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/CanalConnector.java @@ -1,5 +1,6 @@ package com.alibaba.otter.canal.client; +import java.io.IOException; import java.util.concurrent.TimeUnit; import com.alibaba.otter.canal.protocol.Message; diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java b/client/src/main/java/com/alibaba/otter/canal/client/ConsumerBatchMessage.java similarity index 95% rename from client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java rename to client/src/main/java/com/alibaba/otter/canal/client/ConsumerBatchMessage.java index 1def9163..01608f09 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/ConsumerBatchMessage.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/ConsumerBatchMessage.java @@ -1,4 +1,4 @@ -package com.alibaba.otter.canal.client.rocketmq; +package com.alibaba.otter.canal.client; import java.util.List; import java.util.concurrent.CountDownLatch; diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/AliyunCredentialsProvider.java b/client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/AliyunCredentialsProvider.java new file mode 100644 index 00000000..166e196b --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/AliyunCredentialsProvider.java @@ -0,0 +1,54 @@ +/** + * aliyun amqp协议 账号类 + * 暂不支持STS授权情况 + */ +package com.alibaba.otter.canal.client.rabbitmq; + +import com.alibaba.mq.amqp.utils.UserUtils; +import com.rabbitmq.client.impl.CredentialsProvider; +import java.security.InvalidKeyException; +import java.security.NoSuchAlgorithmException; + + +public class AliyunCredentialsProvider implements CredentialsProvider { + + + /** + * Access Key ID + */ + private final String AliyunAccessKey; + + /** + * Access Key Secret + */ + private final String AliyunAccessSecret; + + /** + * 资源主账号ID + */ + private final long resourceOwnerId; + + + public AliyunCredentialsProvider(final String accessKey, final String accessSecret, final long resourceOwnerId) { + this.AliyunAccessKey = accessKey; + this.AliyunAccessSecret = accessSecret; + this.resourceOwnerId = resourceOwnerId; + } + + + @Override + public String getUsername() { + return UserUtils.getUserName(AliyunAccessKey, resourceOwnerId); + } + + @Override + public String getPassword() { + try { + return UserUtils.getPassord(AliyunAccessSecret); + } catch (InvalidKeyException | NoSuchAlgorithmException ignored) { + } + return null; + } + + +} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/RabbitMQCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/RabbitMQCanalConnector.java new file mode 100644 index 00000000..bf43a4a8 --- /dev/null +++ b/client/src/main/java/com/alibaba/otter/canal/client/rabbitmq/RabbitMQCanalConnector.java @@ -0,0 +1,299 @@ +package com.alibaba.otter.canal.client.rabbitmq; + +import com.alibaba.fastjson.JSON; +import com.alibaba.otter.canal.client.CanalMQConnector; +import com.alibaba.otter.canal.client.CanalMessageDeserializer; +import com.alibaba.otter.canal.client.ConsumerBatchMessage; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.protocol.exception.CanalClientException; +import com.google.common.collect.Lists; +import com.rabbitmq.client.*; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +public class RabbitMQCanalConnector implements CanalMQConnector { + + private static final Logger logger = LoggerFactory + .getLogger(RabbitMQCanalConnector.class); + + // 链接地址 + private String nameServer; + + // 主机名 + private String vhost; + + private String queueName; + + // 一些鉴权信息 + private String accessKey; + private String secretKey; + private Long resourceOwnerId; + private String username; + private String password; + + private boolean flatMessage; + + private Connection connect; + private Channel channel; + + + private long batchProcessTimeout = 60 * 1000; + private BlockingQueue messageBlockingQueue; + private volatile ConsumerBatchMessage lastGetBatchMessage = null; + + public RabbitMQCanalConnector(String nameServer, String vhost, String queueName, String accessKey, String secretKey, + String username, String password, Long resourceOwnerId, boolean flatMessage){ + this.nameServer = nameServer; + this.vhost = vhost; + this.queueName = queueName; + this.accessKey = accessKey; + this.secretKey = secretKey; + this.username = username; + this.password = password; + this.resourceOwnerId = resourceOwnerId; + this.flatMessage = flatMessage; + this.messageBlockingQueue = new LinkedBlockingQueue<>(1024); + } + + public void connect() throws CanalClientException { + ConnectionFactory factory = new ConnectionFactory(); + if (accessKey.length() > 0 && secretKey.length() > 0) { + factory.setCredentialsProvider(new AliyunCredentialsProvider(accessKey, secretKey, resourceOwnerId)); + } else { + factory.setUsername(username); + factory.setPassword(password); + } + factory.setHost(nameServer); + factory.setAutomaticRecoveryEnabled(true); + factory.setNetworkRecoveryInterval(5000); + factory.setVirtualHost(vhost); + try { + connect = factory.newConnection(); + channel = connect.createChannel(); + } catch (IOException | TimeoutException e) { + throw new CanalClientException("Start RabbitMQ producer error", e); + } + } + + @Override + public void disconnect() throws CanalClientException { + if (connect != null) { + try { + connect.close(); + } catch (IOException e) { + throw new CanalClientException("stop connect error", e); + } + } + if (channel != null) { + try { + channel.close(); + } catch (IOException | TimeoutException e) { + throw new CanalClientException("stop channel error", e); + } + } + } + + @Override + public boolean checkValid() throws CanalClientException { + return true; // 永远true + } + + /** + * RabbitMQ支持拉取 不需要订阅 + * + * @param filter + * @throws CanalClientException + */ + @Override + public void subscribe(String filter) throws CanalClientException { + // 不存在连接 则重新连接 + if (connect == null) { + this.connect(); + } + + Consumer consumer = new DefaultConsumer(channel) { + + @Override + public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, + byte[] body) throws IOException { + + if (body != null) { + channel.basicAck(envelope.getDeliveryTag(), process(body)); + } + } + }; + try { + channel.basicConsume(queueName, false, consumer); + } catch (IOException e) { + throw new CanalClientException("error", e); + } + } + + @Override + public void subscribe() throws CanalClientException { + this.subscribe(null); + } + + @Override + public void unsubscribe() throws CanalClientException { + // 取消订阅 直接强行断开吧 + this.disconnect(); + } + + @Override + public Message get(int batchSize) throws CanalClientException { + return null; + } + + @Override + public Message get(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + return null; + } + + @Override + public Message getWithoutAck(int batchSize) throws CanalClientException { + throw new CanalClientException("mq not support this method"); + } + + @Override + public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws CanalClientException { + throw new CanalClientException("mq not support this method"); + } + + @Override + public void ack(long batchId) throws CanalClientException { + throw new CanalClientException("mq not support this method"); + } + + @Override + public void rollback(long batchId) throws CanalClientException { + throw new CanalClientException("mq not support this method"); + + } + + @Override + public List getList(Long timeout, TimeUnit unit) throws CanalClientException { + List messages = getListWithoutAck(timeout, unit); + if (messages != null && !messages.isEmpty()) { + ack(); + } + return messages; + } + + @Override + public List getListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException { + try { + if (this.lastGetBatchMessage != null) { + throw new CanalClientException("mq get/ack not support concurrent & async ack"); + } + + ConsumerBatchMessage batchMessage = messageBlockingQueue.poll(timeout, unit); + if (batchMessage != null) { + this.lastGetBatchMessage = batchMessage; + return batchMessage.getData(); + } + } catch (InterruptedException ex) { + logger.warn("Get message timeout", ex); + throw new CanalClientException("Failed to fetch the data after: " + timeout); + } + return Lists.newArrayList(); + } + + @Override + public List getFlatList(Long timeout, TimeUnit unit) throws CanalClientException { + List messages = getFlatListWithoutAck(timeout, unit); + if (messages != null && !messages.isEmpty()) { + ack(); + } + return messages; + } + + @Override + public List getFlatListWithoutAck(Long timeout, TimeUnit unit) throws CanalClientException { + try { + if (this.lastGetBatchMessage != null) { + throw new CanalClientException("mq get/ack not support concurrent & async ack"); + } + + ConsumerBatchMessage batchMessage = messageBlockingQueue.poll(timeout, unit); + if (batchMessage != null) { + this.lastGetBatchMessage = batchMessage; + return batchMessage.getData(); + } + } catch (InterruptedException ex) { + logger.warn("Get message timeout", ex); + throw new CanalClientException("Failed to fetch the data after: " + timeout); + } + return Lists.newArrayList(); + } + + private boolean process(byte[] messageData) { + if (logger.isDebugEnabled()) { + logger.debug("Get Message: {}", new String(messageData)); + } + List messageList = Lists.newArrayList(); + if (!flatMessage) { + Message message = CanalMessageDeserializer.deserializer(messageData); + messageList.add(message); + } else { + FlatMessage flatMessage = JSON.parseObject(messageData, FlatMessage.class); + messageList.add(flatMessage); + } + ConsumerBatchMessage batchMessage; + if (!flatMessage) { + batchMessage = new ConsumerBatchMessage(messageList); + } else { + batchMessage = new ConsumerBatchMessage(messageList); + } + try { + messageBlockingQueue.put(batchMessage); + } catch (InterruptedException e) { + logger.error("Put message to queue error", e); + throw new RuntimeException(e); + } + boolean isCompleted; + try { + isCompleted = batchMessage.waitFinish(batchProcessTimeout); + } catch (InterruptedException e) { + logger.error("Interrupted when waiting messages to be finished.", e); + throw new RuntimeException(e); + } + boolean isSuccess = batchMessage.isSuccess(); + return isCompleted && isSuccess; + } + + @Override + public void ack() throws CanalClientException { + try { + if (this.lastGetBatchMessage != null) { + this.lastGetBatchMessage.ack(); + } + } catch (Throwable e) { + if (this.lastGetBatchMessage != null) { + this.lastGetBatchMessage.fail(); + } + } finally { + this.lastGetBatchMessage = null; + } + } + + @Override + public void rollback() throws CanalClientException { + try { + if (this.lastGetBatchMessage != null) { + this.lastGetBatchMessage.fail(); + } + } finally { + this.lastGetBatchMessage = null; + } + } + +} diff --git a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java index a1a6d83b..71bc5537 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/rocketmq/RocketMQCanalConnector.java @@ -5,6 +5,7 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import com.alibaba.otter.canal.client.ConsumerBatchMessage; import org.apache.commons.lang.StringUtils; import org.apache.rocketmq.acl.common.AclClientRPCHook; import org.apache.rocketmq.acl.common.SessionCredentials; diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java index 54d0199b..49fc3e70 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalConstants.java @@ -69,6 +69,11 @@ public class CanalConstants { public static final String CANAL_MQ_KAFKA_KERBEROS_ENABLE = ROOT + "." + "mq.kafka.kerberos.enable"; public static final String CANAL_MQ_KAFKA_KERBEROS_KRB5FILEPATH = ROOT + "." + "mq.kafka.kerberos.krb5FilePath"; public static final String CANAL_MQ_KAFKA_KERBEROS_JAASFILEPATH = ROOT + "." + "mq.kafka.kerberos.jaasFilePath"; + public static final String CANAL_MQ_USERNAME = ROOT + "." + "mq.username"; + public static final String CANAL_MQ_PASSWORD = ROOT + "." + "mq.password"; + public static final String CANAL_MQ_VHOST = ROOT + "." + "mq.vhost"; + public static final String CANAL_MQ_ALIYUN_UID = ROOT + "." + "mq.aliyunuid"; + public static final String CANAL_MQ_EXCHANGE = ROOT + "." + "mq.exchange"; public static String getInstanceModeKey(String destination) { return MessageFormat.format(INSTANCE_MODE_TEMPLATE, destination); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java index a83b2cb1..38e86221 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalStarter.java @@ -11,6 +11,7 @@ import com.alibaba.otter.canal.common.MQProperties; import com.alibaba.otter.canal.deployer.admin.CanalAdminController; import com.alibaba.otter.canal.kafka.CanalKafkaProducer; import com.alibaba.otter.canal.rocketmq.CanalRocketMQProducer; +import com.alibaba.otter.canal.rabbitmq.CanalRabbitMQProducer; import com.alibaba.otter.canal.server.CanalMQStarter; import com.alibaba.otter.canal.spi.CanalMQProducer; @@ -64,6 +65,8 @@ public class CanalStarter { canalMQProducer = new CanalKafkaProducer(); } else if (serverMode.equalsIgnoreCase("rocketmq")) { canalMQProducer = new CanalRocketMQProducer(); + } else if (serverMode.equalsIgnoreCase("rabbitmq")) { + canalMQProducer = new CanalRabbitMQProducer(); } if (canalMQProducer != null) { @@ -260,6 +263,31 @@ public class CanalStarter { mqProperties.setKerberosJaasFilePath(kafkaKerberosJaasFilepath); } + String vhost = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_VHOST); + if (!StringUtils.isEmpty(vhost)) { + mqProperties.setVhost(vhost); + } + + String username = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_USERNAME); + if (!StringUtils.isEmpty(username)) { + mqProperties.setUsername(username); + } + + String password = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_PASSWORD); + if (!StringUtils.isEmpty(password)) { + mqProperties.setPassword(password); + } + + String aliyunUID = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_ALIYUN_UID); + if (!StringUtils.isEmpty(aliyunUID)) { + mqProperties.setAliyunUID(Long.valueOf(aliyunUID)); + } + + String exchange = CanalController.getProperty(properties, CanalConstants.CANAL_MQ_EXCHANGE); + if (!StringUtils.isEmpty(exchange)) { + mqProperties.setExchange(exchange); + } + for (Object key : properties.keySet()) { key = StringUtils.trim(key.toString()); if (((String) key).startsWith(CanalConstants.CANAL_MQ_PROPERTIES)) { diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index 1a71e42b..a6c938d7 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -130,6 +130,11 @@ canal.mq.producerGroup = test canal.mq.accessChannel = local # aliyun mq namespace #canal.mq.namespace = +canal.mq.vhost= +canal.mq.exchange= +canal.mq.username= +canal.mq.password= +canal.mq.aliyunuid= ################################################## ######### Kafka Kerberos Info ############# diff --git a/deployer/src/main/resources/logback.xml b/deployer/src/main/resources/logback.xml index 024a3af0..a3f66c60 100644 --- a/deployer/src/main/resources/logback.xml +++ b/deployer/src/main/resources/logback.xml @@ -77,7 +77,10 @@ - + + + + diff --git a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java index 4f33d9f6..d277c3a1 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java @@ -15,17 +15,17 @@ import com.alibaba.otter.canal.common.utils.AddressUtils; public class SimpleCanalClientTest extends AbstractCanalClientTest { public SimpleCanalClientTest(String destination){ - super(destination); - } + super(destination); + } public static void main(String args[]) { // 根据ip，直接创建链接，无HA的功能 String destination = "example"; String ip = AddressUtils.getHostIp(); CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(ip, 11111), - destination, - "canal", - "canal"); + destination, + "canal", + "canal"); final SimpleCanalClientTest clientTest = new SimpleCanalClientTest(destination); clientTest.setConnector(connector); diff --git a/pom.xml b/pom.xml index b140ed62..3ac2e2c0 100644 --- a/pom.xml +++ b/pom.xml @@ -101,6 +101,7 @@ true 3.2.18.RELEASE 4.5.2 + 5.5.0 0.8.3 2.22.1 -server -Xms512m -Xmx1024m -Dfile.encoding=UTF-8 @@ -321,6 +322,16 @@ rocketmq-acl ${rocketmq_version} + + com.rabbitmq + amqp-client + ${rabbitmq_version} + + + com.alibaba.mq-amqp + mq-amqp-client + 1.0.3 + javax.annotation javax.annotation-api diff --git a/server/pom.xml b/server/pom.xml index ced10081..d9926645 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -45,6 +45,14 @@ org.apache.rocketmq rocketmq-acl + + com.rabbitmq + amqp-client + + + com.alibaba.mq-amqp + mq-amqp-client + org.jboss.netty diff --git a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java index c147c4c5..1304b859 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java @@ -33,6 +33,11 @@ public class MQProperties { private boolean kerberosEnable = false; // kafka集群是否启动Kerberos认证 private String kerberosKrb5FilePath = ""; // 启动Kerberos认证时配置为krb5.conf文件的路径 private String kerberosJaasFilePath = ""; // 启动Kerberos认证时配置为jaas.conf文件的路径 + private String username = ""; // rabbitmq 账号 + private String password = ""; // rabbitmq 密码 + private String vhost = ""; // rabbitmq 密码 + private long aliyunUID = 0; // aliyun 用户ID rabbitmq 阿里云需要使用 + private String exchange = ""; // rabbitmq 交换机 public static class CanalDestination { @@ -276,6 +281,46 @@ public class MQProperties { this.kerberosJaasFilePath = kerberosJaasFilePath; } + public void setPassword(String password) { + this.password = password; + } + + public String getPassword() { + return password; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getUsername() { + return username; + } + + public void setVhost(String vhost) { + this.vhost = vhost; + } + + public String getVhost() { + return vhost; + } + + public long getAliyunUID() { + return aliyunUID; + } + + public void setAliyunUID(long aliyunUID) { + this.aliyunUID = aliyunUID; + } + + public String getExchange() { + return exchange; + } + + public void setExchange(String exchange) { + this.exchange = exchange; + } + @Override public String toString() { return "MQProperties{" + "servers='" + servers + '\'' + ", retries=" + retries + ", batchSize=" + batchSize diff --git a/server/src/main/java/com/alibaba/otter/canal/rabbitmq/AliyunCredentialsProvider.java b/server/src/main/java/com/alibaba/otter/canal/rabbitmq/AliyunCredentialsProvider.java new file mode 100644 index 00000000..279dffa9 --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/rabbitmq/AliyunCredentialsProvider.java @@ -0,0 +1,55 @@ +/** + * aliyun amqp协议 账号类 + * 暂不支持STS授权情况 + */ +package com.alibaba.otter.canal.rabbitmq; + +import java.security.InvalidKeyException; +import java.security.NoSuchAlgorithmException; + +import com.alibaba.mq.amqp.utils.UserUtils; +import com.rabbitmq.client.impl.CredentialsProvider; + + +public class AliyunCredentialsProvider implements CredentialsProvider { + + + /** + * Access Key ID + */ + private final String AliyunAccessKey; + + /** + * Access Key Secret + */ + private final String AliyunAccessSecret; + + /** + * 资源主账号ID + */ + private final long resourceOwnerId; + + + public AliyunCredentialsProvider(final String accessKey, final String accessSecret, final long resourceOwnerId) { + this.AliyunAccessKey = accessKey; + this.AliyunAccessSecret = accessSecret; + this.resourceOwnerId = resourceOwnerId; + } + + + @Override + public String getUsername() { + return UserUtils.getUserName(AliyunAccessKey, resourceOwnerId); + } + + @Override + public String getPassword() { + try { + return UserUtils.getPassord(AliyunAccessSecret); + } catch (InvalidKeyException | NoSuchAlgorithmException ignored) { + } + return null; + } + + +} diff --git a/server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java new file mode 100644 index 00000000..06c8a775 --- /dev/null +++ b/server/src/main/java/com/alibaba/otter/canal/rabbitmq/CanalRabbitMQProducer.java @@ -0,0 +1,119 @@ +package com.alibaba.otter.canal.rabbitmq; + +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.serializer.SerializerFeature; +import com.alibaba.otter.canal.common.CanalMessageSerializer; +import com.alibaba.otter.canal.common.MQMessageUtils; +import com.alibaba.otter.canal.common.MQProperties; +import com.alibaba.otter.canal.protocol.FlatMessage; +import com.alibaba.otter.canal.protocol.Message; +import com.alibaba.otter.canal.server.exception.CanalServerException; +import com.alibaba.otter.canal.spi.CanalMQProducer; +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; +import com.rabbitmq.client.ConnectionFactory; + +import org.apache.commons.lang.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeoutException; + +public class CanalRabbitMQProducer implements CanalMQProducer { + + private static final Logger logger = LoggerFactory.getLogger(CanalRabbitMQProducer.class); + private MQProperties mqProperties; + private Connection connect; + private Channel channel; + + @Override + public void init(MQProperties mqProperties) { + this.mqProperties = mqProperties; + ConnectionFactory factory = new ConnectionFactory(); + factory.setHost(mqProperties.getServers()); + if (mqProperties.getAliyunAccessKey().length() > 0 && mqProperties.getAliyunSecretKey().length() > 0 + && mqProperties.getAliyunUID() > 0) { + factory.setCredentialsProvider(new AliyunCredentialsProvider(mqProperties.getAliyunAccessKey(), + mqProperties.getAliyunSecretKey(), + mqProperties.getAliyunUID())); + } else { + factory.setUsername(mqProperties.getUsername()); + factory.setPassword(mqProperties.getPassword()); + } + factory.setVirtualHost(mqProperties.getVhost()); + try { + connect = factory.newConnection(); + channel = connect.createChannel(); + // channel.exchangeDeclare(mqProperties.getExchange(), "topic"); + } catch (IOException | TimeoutException ex) { + throw new CanalServerException("Start RabbitMQ producer error", ex); + } + } + + @Override + public void send(MQProperties.CanalDestination canalDestination, Message message, + Callback callback) throws IOException { + try { + if (!StringUtils.isEmpty(canalDestination.getDynamicTopic())) { + // 动态topic + Map messageMap = MQMessageUtils + .messageTopics(message, canalDestination.getTopic(), canalDestination.getDynamicTopic()); + + for (Map.Entry entry : messageMap.entrySet()) { + String topicName = entry.getKey().replace('.', '_'); + com.alibaba.otter.canal.protocol.Message messageSub = entry.getValue(); + send(canalDestination, topicName, messageSub); + } + } else { + send(canalDestination, canalDestination.getTopic(), message); + } + callback.commit(); + } catch (Throwable e) { + callback.rollback(); + } + } + + private void send(MQProperties.CanalDestination canalDestination, String topicName, + Message messageSub) throws Exception { + if (!mqProperties.getFlatMessage()) { + byte[] message = CanalMessageSerializer.serializer(messageSub, mqProperties.isFilterTransactionEntry()); + if (logger.isDebugEnabled()) { + logger.debug("send message:{} to destination:{}", message, canalDestination.getCanalDestination()); + } + sendMessage(topicName, message); + } else { + List flatMessages = MQMessageUtils.messageConverter(messageSub); + if (flatMessages != null) { + for (FlatMessage flatMessage : flatMessages) { + byte[] message = JSON.toJSONString(flatMessage, SerializerFeature.WriteMapNullValue).getBytes(); + if (logger.isDebugEnabled()) { + logger.debug("send message:{} to destination:{}", + message, + canalDestination.getCanalDestination()); + } + sendMessage(topicName, message); + } + } + } + + } + + private void sendMessage(String queueName, byte[] message) throws Exception { + // tips: 目前逻辑中暂不处理对exchange处理，请在Console后台绑定 才可使用routekey + channel.basicPublish(mqProperties.getExchange(), queueName, null, message); + } + + @Override + public void stop() { + logger.info("## Stop RabbitMQ producer##"); + try { + this.connect.close(); + this.channel.close(); + } catch (IOException | TimeoutException ex) { + throw new CanalServerException("Stop RabbitMQ producer error", ex); + } + } +}