Просмотр исходного кода

fix aliyunrocketmq bug

release_toaliyun_real
xiaohanzi 6 лет назад
Родитель
Сommit
16c93a833e
3 измененных файлов: 53 добавлений и 33 удалений
  1. +13
    -15
      mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java
  2. +13
    -3
      mallinkService/src/main/java/com/iformall/enums/EnumMsgMqTopic.java
  3. +27
    -15
      mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java

+ 13
- 15
mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java Просмотреть файл

@@ -43,22 +43,20 @@ public class AliyunRocketMqConsumer extends MqBaseConsumer {
@PostConstruct
public void init() {
Properties properties = new Properties();
// 您在控制台创建的 Group ID
properties.put(PropertyKeyConst.GROUP_ID, groupId);
// AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.AccessKey, accessKeyId);
// SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, accessKeySecret);
// 设置 TCP 接入域名,到控制台的实例基本信息中查看
properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr);
// 集群订阅方式(默认)
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
// 广播订阅方式
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
for (EnumMsgMqTopic topic: EnumMsgMqTopic.values()) {
Properties properties = new Properties();
// 您在控制台创建的 Group ID
properties.put(PropertyKeyConst.GROUP_ID, topic.getGroupId());
// AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.AccessKey, accessKeyId);
// SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, accessKeySecret);
// 设置 TCP 接入域名,到控制台的实例基本信息中查看
properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr);
// 集群订阅方式(默认)
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
// 广播订阅方式
// properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);
Consumer consumer = ONSFactory.createConsumer(properties);
consumer.subscribe(topic.getCode(), "*", new MessageListener() { //订阅多个 Tag
public Action consume(Message message, ConsumeContext context) {


+ 13
- 3
mallinkService/src/main/java/com/iformall/enums/EnumMsgMqTopic.java Просмотреть файл

@@ -4,8 +4,8 @@ package com.iformall.enums;
* Created by luozukai
*/
public enum EnumMsgMqTopic {
DEFAULT("topic-1", "默认"),
STOCK("stock","库存"),
DEFAULT("topic-1","GID_P_1", "默认"),
STOCK("stock","GID_P_2","库存"),
;

public static EnumMsgMqTopic getEnum(String code) {
@@ -19,10 +19,12 @@ public enum EnumMsgMqTopic {

private String code;
private String message;
private String groupId;

EnumMsgMqTopic(String code, String message) {
EnumMsgMqTopic(String code, String groupId,String message) {
this.code = code;
this.message = message;
this.groupId = groupId;
}

public String getCode() {
@@ -32,4 +34,12 @@ public enum EnumMsgMqTopic {
public String getMessage() {
return message;
}

public String getGroupId() {
return groupId;
}

public void setGroupId(String groupId) {
this.groupId = groupId;
}
}

+ 27
- 15
mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java Просмотреть файл

@@ -1,6 +1,8 @@
package com.iformall.mq.impl;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import javax.annotation.PostConstruct;
@@ -18,6 +20,7 @@ import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.iformall.domain.po.msg.BaseMsg;
import com.iformall.enums.EnumMsgMqTopic;
import com.iformall.mq.MQConfig;
import com.iformall.mq.MqBaseProducer;
import com.iformall.utils.JsonUtil;
@@ -52,28 +55,37 @@ public class AliyunRocketMqMessageProducer implements MqBaseProducer {
//@Autowired
//private WxMsgRecordService wxMsgRecordService;

private static Producer producer = null;
private static Map<String,Producer> producerMap = null;
@PostConstruct
public void init() {
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.GROUP_ID, groupId);
// AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.AccessKey,accessKeyId);
// SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, accessKeySecret);
//设置发送超时时间,单位毫秒
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
// 设置 TCP 接入域名,到控制台的实例基本信息中查看
properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr);
producer = ONSFactory.createProducer(properties);
// 在发送消息前,必须调用 start 方法来启动 Producer,只需调用一次即可
producer.start();
log.info("aliyunrocketmq producer info :"+JSON.toJSONString(producer));
producerMap = new HashMap<String,Producer>();
for (EnumMsgMqTopic topic: EnumMsgMqTopic.values()) {
Properties properties = new Properties();
properties.setProperty(PropertyKeyConst.GROUP_ID, topic.getGroupId());
// AccessKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.AccessKey,accessKeyId);
// SecretKey 阿里云身份验证,在阿里云服务器管理控制台创建
properties.put(PropertyKeyConst.SecretKey, accessKeySecret);
//设置发送超时时间,单位毫秒
properties.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, "3000");
// 设置 TCP 接入域名,到控制台的实例基本信息中查看
properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr);
Producer producer = ONSFactory.createProducer(properties);
// 在发送消息前,必须调用 start 方法来启动 Producer,只需调用一次即可
producer.start();
log.info("aliyunrocketmq producer info :"+JSON.toJSONString(producer));
producerMap.put(topic.getCode(), producer);
}
}
@Override
public void sendMessage(BaseMsg data, String topic, String tags, String keys) {
Producer producer = producerMap.get(topic);
if (null == producer) {
log.error("topic :"+topic+" has no AliyunRocketMqMessageProducer.");
return;
}
if (producer.isClosed()) {
producer.start();
}


Загрузка…
Отмена
Сохранить