阿里云ONS RocketMQ接入指南

1. 配置阿里云accessKey

@Configuration
@ConfigurationProperties(prefix = "aliyun.auth")
public class AliyunAuthConfig {private String accessKey;private String secretKey;public String getAccessKey() {return accessKey;}public void setAccessKey(String accessKey) {this.accessKey = accessKey;}public String getSecretKey() {return secretKey;}public void setSecretKey(String secretKey) {this.secretKey = secretKey;}
}

2. 注册ProducerBean

@Configuration
public class RocketMqProducerConfiguration {@Autowiredprivate AliyunAuthConfig mqConfig;@Value("${rocketmq.server.address}")private String nameSrvAddr;@Bean(initMethod = "start", destroyMethod = "shutdown")public ProducerBean buildProducer() {ProducerBean producer = new ProducerBean();producer.setProperties(getMqProperties());return producer;}public Properties getMqProperties() {Properties properties = new Properties();properties.setProperty(PropertyKeyConst.AccessKey, mqConfig.getAccessKey());properties.setProperty(PropertyKeyConst.SecretKey, mqConfig.getSecretKey());properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr);return properties;}}

3. 注册Consumer

@Configuration
public class RocketMqConsumerConfiguration {@Value("${rocketmq.server.address}")private String nameSrvAddr;private final String GID = "GID_FIRST_CHOICE_COOP_ORDER";@Autowiredprivate AliyunAuthConfig aliyunAuthConfig;@Autowired(required = false)private List<MqSubscriptionConfig> mqSubscriptionConfigList;@Bean(initMethod = "start", destroyMethod = "shutdown", name = "consumerBean")public ConsumerBean buildConsumer() {ConsumerBean consumerBean = new ConsumerBean();//配置文件Properties properties = getProperties();properties.setProperty(PropertyKeyConst.AccessKey, aliyunAuthConfig.getAccessKey());properties.setProperty(PropertyKeyConst.SecretKey, aliyunAuthConfig.getSecretKey());properties.setProperty(PropertyKeyConst.GROUP_ID, GID);//将消费者线程数固定为20个 20为默认值properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");consumerBean.setProperties(properties);//订阅关系Map<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>(1);if(!CollectionUtils.isEmpty(mqSubscriptionConfigList)){for (MqSubscriptionConfig mqSubscriptionConfig : mqSubscriptionConfigList) {Subscription subscription = new Subscription();subscription.setTopic(mqSubscriptionConfig.getTopic());subscription.setExpression(mqSubscriptionConfig.getExpression());subscriptionTable.put(subscription, mqSubscriptionConfig.getMessageListener());}}consumerBean.setSubscriptionTable(subscriptionTable);return consumerBean;}private Properties getProperties() {Properties properties = new Properties();properties.setProperty(PropertyKeyConst.AccessKey, aliyunAuthConfig.getAccessKey());properties.setProperty(PropertyKeyConst.SecretKey, aliyunAuthConfig.getSecretKey());properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr);return properties;}
}

多个consumer可以批量注册

@AllArgsConstructor
@Data
public class MqSubscriptionConfig {private String Topic;private String expression;private MessageListener messageListener;}

@Beanpublic MqSubscriptionConfig cooperationPaymentOrderConfig(){return new MqSubscriptionConfig("TOPIC_PAYMENT_ORDER", "*", cooperationPaymentOrderListener);}@Beanpublic MqSubscriptionConfig cooperationSettlementOrderConfig(){return new MqSubscriptionConfig("TOPIC_SETTLEMENT_ORDER", "*", cooperationSettlementOrderListener);}@Beanpublic MqSubscriptionConfig captainPromotionLinkConfig(){return new MqSubscriptionConfig("TOPIC_CAPTAIN_PROMOTION_LINK", "*", captainPromotionLinkListener);}public class PaymentOrderListener implements MessageListener {}

阿里云ONS RocketMQ接入指南相关推荐

JAVA整合阿里云ONS(RocketMQ)
前言关于阿里云ONS我这里不多介绍,用的话直接去看官网,这里提几个实际对接的问题问题一: TCP版的不支持本地调试,只支持公网链接问题二: ONS不支持批量消息问题三: ONS延迟消息的时间是 ...
阿里云ONS / RocketMQ的定时消息 / 延时消息
考虑延时和定时消息,是因为遇到了一个业务场景: 前置任务完成时发送消息,但因为一些业务原因,不希望消息马上被消费,因此需要设置延时. 文章目录几种解决思路实现方案 ONS延迟消息 RocketMQ ...
【结果很简单,过程很艰辛】记阿里云Ons消息队列服务.NET接口填坑过程
Maybe 这个问题很简单,因为解决方法是非常简单,但填坑过程会把人逼疯,在阿里云ONS工作人员.同事和朋友的协助下,经过一天的调试和瞎捣鼓,终于解决了这个坑,把问题记下来,也许更多人在碰到类似问题的 ...
阿里云服务器新手选型指南
阿里云服务器新手选型指南: 小型网站:1核2G共享型以上中大型网站:4核8G通用型以上项目临时测试:突发性能型.共享型,配置够用就行. 更多应用,比如分布式计算.AI学习等等,则需选择更好的云服务 ...
BC26电信云平台LWM2M接入指南
BC26电信云平台LWM2M接入指南最近使用移远BC26做了电信云平台的LWM2M介入设计,遇到几个问题,逐一解决,做个记录准备工作相对于移动的OneNet平台,电信平台的LWM2M接入指令完全 ...
阿里云服务器安装RocketMQ及其控制台
1. 环境说明本文所使用的阿里云服务器为CentOS8,JDK为1.8,Maven为3.6.3,RocketMQ为4.9.1 2. 下载安装包进入RocketMQ官网点击最新版本,进入点击下载 ...
阿里云网站应用支付接入
阿里云网站应用支付接入 [本文使用沙箱环境进行开发测试] 一.配置沙箱环境 1.进入[支付宝开放平台],找到"研发服务"查看沙箱应用信息 2.设置沙箱应用的公钥.密钥二.引入依赖 ...
阿里云ONS消息队列入门指南
文章目录概述 springboot整合接入 mq消费者 mq 生产者概述消息队列 RocketMQ 是阿里巴巴集团基于高可用分布式集群技术,自主研发的云正式商用的专业消息中间件,既可为分布式应用 ...

阿里云ONS RocketMQ接入指南

阿里云ONS RocketMQ接入指南相关推荐

最新文章

热门文章