From 5d7d19fdb1a4e94f3347ae91f64e635a395b532e Mon Sep 17 00:00:00 2001 From: Smile-cmd <62556000+Smile-cmd@users.noreply.github.com> Date: Thu, 16 Mar 2023 18:42:37 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0rabbitmq=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=EF=BC=8C=E4=BB=A5=E6=8C=81=E4=B9=85=E5=8C=96=E6=96=B9=E5=BC=8F?= =?UTF-8?q?=E6=8A=95=E9=80=92=E6=B6=88=E6=81=AF=20(#4644)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: wangheng <760261296@qq.com> --- .../rabbitmq/config/RabbitMQConstants.java | 3 +++ .../config/RabbitMQProducerConfig.java | 27 +++++++++++++++++++ .../producer/CanalRabbitMQProducer.java | 24 ++++++++++++----- 3 files changed, 48 insertions(+), 6 deletions(-) diff --git a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQConstants.java b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQConstants.java index f6b6f8e6..1a0b72b4 100644 --- a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQConstants.java +++ b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQConstants.java @@ -15,6 +15,9 @@ public class RabbitMQConstants { public static final String RABBITMQ_VIRTUAL_HOST = ROOT + "." + "virtual.host"; public static final String RABBITMQ_USERNAME = ROOT + "." + "username"; public static final String RABBITMQ_PASSWORD = ROOT + "." + "password"; + public static final String RABBITMQ_QUEUE = ROOT + "." + "queue"; + public static final String RABBITMQ_ROUTING_KEY = ROOT + "." + "routingKey"; + public static final String RABBITMQ_DELIVERY_MODE = ROOT + "." + "deliveryMode"; public static final String RABBITMQ_RESOURCE_OWNERID = ROOT + "." + "rabbitmq.resource.ownerId"; } diff --git a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQProducerConfig.java b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQProducerConfig.java index 57a9d12d..8dcde5ca 100644 --- a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQProducerConfig.java +++ b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/config/RabbitMQProducerConfig.java @@ -15,6 +15,9 @@ public class RabbitMQProducerConfig extends MQProperties { private String exchange; private String username; private String password; + private String queue; + private String routingKey; + private String deliveryMode; public String getHost() { return host; @@ -55,4 +58,28 @@ public class RabbitMQProducerConfig extends MQProperties { public void setPassword(String password) { this.password = password; } + + public String getQueue() { + return queue; + } + + public void setQueue(String queue) { + this.queue = queue; + } + + public String getRoutingKey() { + return routingKey; + } + + public void setRoutingKey(String routingKey) { + this.routingKey = routingKey; + } + + public String getDeliveryMode() { + return deliveryMode; + } + + public void setDeliveryMode(String deliveryMode) { + this.deliveryMode = deliveryMode; + } } diff --git a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java index 80f70854..f5bfddb7 100644 --- a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java +++ b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java @@ -6,6 +6,7 @@ import java.util.Map; import java.util.Properties; import java.util.concurrent.TimeoutException; +import com.rabbitmq.client.*; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -26,10 +27,6 @@ import com.alibaba.otter.canal.connector.rabbitmq.config.RabbitMQConstants; import com.alibaba.otter.canal.connector.rabbitmq.config.RabbitMQProducerConfig; import com.alibaba.otter.canal.protocol.FlatMessage; import com.alibaba.otter.canal.protocol.Message; -import com.rabbitmq.client.AlreadyClosedException; -import com.rabbitmq.client.Channel; -import com.rabbitmq.client.Connection; -import com.rabbitmq.client.ConnectionFactory; /** * RabbitMQ Producer SPI 实现 @@ -75,7 +72,10 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ try { connect = factory.newConnection(); channel = connect.createChannel(); - // channel.exchangeDeclare(mqProperties.getExchange(), "topic"); + channel.queueDeclare(rabbitMQProperties.getQueue(), true, false, false, null); + channel.exchangeDeclare(rabbitMQProperties.getExchange(), rabbitMQProperties.getDeliveryMode(), true, false, false, null); + channel.queueBind(rabbitMQProperties.getQueue(), rabbitMQProperties.getExchange(), rabbitMQProperties.getRoutingKey()); + } catch (IOException | TimeoutException ex) { throw new CanalException("Start RabbitMQ producer error", ex); } @@ -106,6 +106,18 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ if (!StringUtils.isEmpty(password)) { rabbitMQProperties.setPassword(password); } + String queue = PropertiesUtils.getProperty(properties, RabbitMQConstants.RABBITMQ_QUEUE); + if (!StringUtils.isEmpty(queue)) { + rabbitMQProperties.setQueue(queue); + } + String routingKey = PropertiesUtils.getProperty(properties, RabbitMQConstants.RABBITMQ_ROUTING_KEY); + if (!StringUtils.isEmpty(routingKey)) { + rabbitMQProperties.setRoutingKey(routingKey); + } + String deliveryMode = PropertiesUtils.getProperty(properties, RabbitMQConstants.RABBITMQ_DELIVERY_MODE); + if (!StringUtils.isEmpty(deliveryMode)) { + rabbitMQProperties.setDeliveryMode(deliveryMode); + } } @Override @@ -165,7 +177,7 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ // tips: 目前逻辑中暂不处理对exchange处理，请在Console后台绑定 才可使用routekey try { RabbitMQProducerConfig rabbitMQProperties = (RabbitMQProducerConfig) this.mqProperties; - channel.basicPublish(rabbitMQProperties.getExchange(), queueName, null, message); + channel.basicPublish(rabbitMQProperties.getExchange(), queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message); } catch (Throwable e) { throw new RuntimeException(e); }