chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

Support client connect rocketmq in aliyun

Browse Source
This commit is contained in:
maowei.ymw
2018-11-19 14:33:59 +08:00
parent 7992b883d2
commit 50630cc934
6 changed files with 78 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
server/src/main/java/com/alibaba/otter/canal/common/MQProperties.java
View File

@@ -22,6 +22,8 @@ public class MQProperties {
private boolean flatMessage = true;
private String compressionType = "none";
private String acks = "all";
private String aliyunAccessKey = "";
private String aliyunSecretKey = "";
public static class CanalDestination {
@@ -168,4 +170,19 @@ public class MQProperties {
this.acks = acks;
}
public String getAliyunAccessKey() {
return aliyunAccessKey;
}
public void setAliyunAccessKey(String aliyunAccessKey) {
this.aliyunAccessKey = aliyunAccessKey;
}
public String getAliyunSecretKey() {
return aliyunSecretKey;
}
public void setAliyunSecretKey(String aliyunSecretKey) {
this.aliyunSecretKey = aliyunSecretKey;
}
}

16
server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
View File

@@ -7,12 +7,15 @@ import com.alibaba.otter.canal.common.MQProperties;
import com.alibaba.otter.canal.protocol.FlatMessage;
import com.alibaba.otter.canal.server.exception.CanalServerException;
import com.alibaba.otter.canal.spi.CanalMQProducer;
import com.aliyun.openservices.apache.api.impl.authority.SessionCredentials;
import com.aliyun.openservices.apache.api.impl.rocketmq.ClientRPCHook;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -30,10 +33,19 @@ public class CanalRocketMQProducer implements CanalMQProducer {
@Override
public void init(MQProperties rocketMQProperties) {
this.mqProperties = rocketMQProperties;
defaultMQProducer = new DefaultMQProducer();
RPCHook rpcHook = null;
if(rocketMQProperties.getAliyunAccessKey().length() > 0
&& rocketMQProperties.getAliyunSecretKey().length() > 0){
SessionCredentials sessionCredentials = new SessionCredentials();
sessionCredentials.setAccessKey(rocketMQProperties.getAliyunAccessKey());
sessionCredentials.setSecretKey(rocketMQProperties.getAliyunSecretKey());
rpcHook = new ClientRPCHook(sessionCredentials);
}
defaultMQProducer = new DefaultMQProducer(rocketMQProperties.getProducerGroup(), rpcHook);
defaultMQProducer.setNamesrvAddr(rocketMQProperties.getServers());
defaultMQProducer.setProducerGroup(rocketMQProperties.getProducerGroup());
defaultMQProducer.setRetryTimesWhenSendFailed(rocketMQProperties.getRetries());
defaultMQProducer.setVipChannelEnabled(false);
logger.info("##Start RocketMQ producer##");
try {
defaultMQProducer.start();