diff --git a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java index c4eb3d6d5..c30d5f807 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java +++ b/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) { diff --git a/mallinkService/src/main/java/com/iformall/enums/EnumMsgMqTopic.java b/mallinkService/src/main/java/com/iformall/enums/EnumMsgMqTopic.java index e342de989..55cd95f91 100644 --- a/mallinkService/src/main/java/com/iformall/enums/EnumMsgMqTopic.java +++ b/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; + } } diff --git a/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java b/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java index 9ef3d3c11..e539e831d 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java +++ b/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 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(); + 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(); }