From 75174a63afed37d2569b7f4930dec3c47923c597 Mon Sep 17 00:00:00 2001 From: 995586041 <995586041@qq.com> Date: Mon, 19 Apr 2021 10:29:23 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0rabbitmq=E6=B6=88=E6=81=AF?= =?UTF-8?q?=E6=8C=81=E4=B9=85=E5=8C=96deliveryMode=E9=85=8D=E7=BD=AE=20(#3?= =?UTF-8?q?300)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../rabbitmq/config/RabbitMQConstants.java | 1 + .../rabbitmq/config/RabbitMQProducerConfig.java | 10 ++++++++++ .../rabbitmq/producer/CanalRabbitMQProducer.java | 13 ++++++++----- deployer/src/main/resources/canal.properties | 5 +++-- 4 files changed, 22 insertions(+), 7 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..e52183bd 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,7 @@ 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_DELIVERY_NODE = 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..7db4763d 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,8 @@ public class RabbitMQProducerConfig extends MQProperties { private String exchange; private String username; private String password; + // 1:transient, 2:"persistent" + private String deliveryMode; public String getHost() { return host; @@ -55,4 +57,12 @@ public class RabbitMQProducerConfig extends MQProperties { public void setPassword(String password) { this.password = password; } + + 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 07596586..274023a9 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 实现 @@ -106,6 +103,10 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ if (!StringUtils.isEmpty(password)) { rabbitMQProperties.setPassword(password); } + String deliveryMode = properties.getProperty(RabbitMQConstants.RABBITMQ_DELIVERY_NODE); + if (!StringUtils.isEmpty(deliveryMode)) { + rabbitMQProperties.setDeliveryMode(deliveryMode); + } } @Override @@ -165,7 +166,9 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ // tips: 目前逻辑中暂不处理对exchange处理，请在Console后台绑定 才可使用routekey try { RabbitMQProducerConfig rabbitMQProperties = (RabbitMQProducerConfig) this.mqProperties; - channel.basicPublish(rabbitMQProperties.getExchange(), queueName, null, message); + AMQP.BasicProperties properties = rabbitMQProperties != null && rabbitMQProperties.getDeliveryMode() != null + && "2".equals(rabbitMQProperties.getDeliveryMode()) ? MessageProperties.MINIMAL_PERSISTENT_BASIC : null; + channel.basicPublish(rabbitMQProperties.getExchange(), queueName, properties, message); } catch (Throwable e) { throw new RuntimeException(e); } diff --git a/deployer/src/main/resources/canal.properties b/deployer/src/main/resources/canal.properties index bb7b9f2b..a60afd56 100644 --- a/deployer/src/main/resources/canal.properties +++ b/deployer/src/main/resources/canal.properties @@ -2,7 +2,7 @@ ######### common argument ############# ################################################# # tcp bind ip -canal.ip = +canal.ip = # register ip to zookeeper canal.register.ip = canal.port = 11111 @@ -166,4 +166,5 @@ rabbitmq.host = rabbitmq.virtual.host = rabbitmq.exchange = rabbitmq.username = -rabbitmq.password = \ No newline at end of file +rabbitmq.password = +rabbitmq.deliveryMode = \ No newline at end of file