From be3a0197d82b7010fb3abdc7e90390e41474931e Mon Sep 17 00:00:00 2001 From: xiaohanzi Date: Sun, 26 Sep 2021 19:28:10 +0800 Subject: [PATCH] fix aliyunrocketmq --- common/pom.xml | 10 ++ .../smart/pos/mq/AliyunRocketMqConsumer.java | 75 ++++++++++ .../pos/mq/AliyunRocketMqMessageProducer.java | 118 ++++++++++++++++ .../neusoft/smart/pos/mq/EnumMsgMqTopic.java | 45 ++++++ .../com/neusoft/smart/pos/mq/JsonUtil.java | 128 ++++++++++++++++++ .../com/neusoft/smart/pos/mq/MQConfig.java | 9 ++ .../com/neusoft/smart/pos/mq/MqProducer.java | 7 + 7 files changed, 392 insertions(+) create mode 100644 common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqConsumer.java create mode 100644 common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqMessageProducer.java create mode 100644 common/src/main/java/com/neusoft/smart/pos/mq/EnumMsgMqTopic.java create mode 100644 common/src/main/java/com/neusoft/smart/pos/mq/JsonUtil.java create mode 100644 common/src/main/java/com/neusoft/smart/pos/mq/MQConfig.java create mode 100644 common/src/main/java/com/neusoft/smart/pos/mq/MqProducer.java diff --git a/common/pom.xml b/common/pom.xml index 389cc0a..71971dd 100644 --- a/common/pom.xml +++ b/common/pom.xml @@ -20,6 +20,16 @@ framework 1.0-SNAPSHOT + + net.sf.dozer + dozer + 5.5.1 + + + com.aliyun.openservices + ons-client + 1.8.4.Final + mysql diff --git a/common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqConsumer.java b/common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqConsumer.java new file mode 100644 index 0000000..fc11b86 --- /dev/null +++ b/common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqConsumer.java @@ -0,0 +1,75 @@ +package com.neusoft.smart.pos.mq; + +import java.util.Properties; + +import javax.annotation.PostConstruct; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Service; + +import com.alibaba.fastjson.JSON; +import com.aliyun.openservices.ons.api.Action; +import com.aliyun.openservices.ons.api.ConsumeContext; +import com.aliyun.openservices.ons.api.Consumer; +import com.aliyun.openservices.ons.api.Message; +import com.aliyun.openservices.ons.api.MessageListener; +import com.aliyun.openservices.ons.api.ONSFactory; +import com.aliyun.openservices.ons.api.PropertyKeyConst; +import com.aliyun.openservices.ons.api.bean.ConsumerBean; + +import lombok.extern.slf4j.Slf4j; + +@Slf4j +@Service +@Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ) +public class AliyunRocketMqConsumer extends MqBaseConsumer { + + @Value("${spring.aliyunRocketmq.accessKeyId}") + private String accessKeyId; + + @Value("${spring.aliyunRocketmq.accessKeySecret}") + private String accessKeySecret; + + @Value("${spring.aliyunRocketmq.groupId}") + private String groupId; + + @Value("${spring.aliyunRocketmq.namesrvAddr}") + private String namesrvAddr; + + @PostConstruct + public void init() { + 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) { + try { + System.out.println("Receive: " + message); + doMessage(new String(message.getBody())); + return Action.CommitMessage; + }catch(Exception e) { + log.error("AliyunRocketMqConsumer error:",e); + return Action.ReconsumeLater; + } + } + }); + consumer.start(); + log.info(" aliyunrocketMq consumer start success! "+JSON.toJSONString(consumer)); + } + + } + +} diff --git a/common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqMessageProducer.java b/common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqMessageProducer.java new file mode 100644 index 0000000..b66519f --- /dev/null +++ b/common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqMessageProducer.java @@ -0,0 +1,118 @@ +package com.neusoft.smart.pos.mq; + +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; + +import javax.annotation.PostConstruct; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Profile; +import org.springframework.stereotype.Service; + +import com.alibaba.fastjson.JSON; +import com.aliyun.openservices.ons.api.Message; +import com.aliyun.openservices.ons.api.ONSFactory; +import com.aliyun.openservices.ons.api.Producer; +import com.aliyun.openservices.ons.api.PropertyKeyConst; +import com.aliyun.openservices.ons.api.SendResult; + + +/** + * @Auther: zfy 阿里云rocketmq + * @Date: 2019-01-31 + * @Description: RocketMQ消息生产者 + */ +@Service +@Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ) +public class AliyunRocketMqMessageProducer implements MqProducer { + private final Logger log = LoggerFactory.getLogger(this.getClass()); + + + @Value("${spring.aliyunRocketmq.accessKeyId}") + private String accessKeyId; + + @Value("${spring.aliyunRocketmq.accessKeySecret}") + private String accessKeySecret; + + @Value("${spring.aliyunRocketmq.groupId}") + private String groupId; + + @Value("${spring.aliyunRocketmq.namesrvAddr}") + private String namesrvAddr; + + //@Autowired + //private WxMsgRecordMapper wxMsgRecordMapper; + + //@Autowired + //private WxMsgRecordService wxMsgRecordService; + + private static Map producerMap = null; + + @PostConstruct + public void init() { + 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(Object 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(); + } + //循环发送消息 + Message msg = new Message( // + // Message 所属的 Topic + topic, + // Message Tag 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在 MQ 服务器过滤 + tags, + // Message Body 可以是任何二进制形式的数据, MQ 不做任何干预, + // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式 + JsonUtil.obj2Json(data).getBytes()); + // 设置代表消息的业务关键属性,请尽可能全局唯一。 + // 以方便您在无法正常收到消息情况下,可通过阿里云服务器管理控制台查询消息并补发 + // 注意:不设置也不会影响消息正常收发 + msg.setKey(keys); + + try { + SendResult sendResult = producer.send(msg); + // 同步发送消息,只要不抛异常就是成功 + if (sendResult != null) { + System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId()); + } + } + catch (Exception e) { + // 消息发送失败,需要进行重试处理,可重新发送这条消息或持久化这条数据进行补偿处理 + System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic()); + e.printStackTrace(); + } + + // 在应用退出前,销毁 Producer 对象 + // 注意:如果不销毁也没有问题 + //producer.shutdown(); + } +} diff --git a/common/src/main/java/com/neusoft/smart/pos/mq/EnumMsgMqTopic.java b/common/src/main/java/com/neusoft/smart/pos/mq/EnumMsgMqTopic.java new file mode 100644 index 0000000..487c8c2 --- /dev/null +++ b/common/src/main/java/com/neusoft/smart/pos/mq/EnumMsgMqTopic.java @@ -0,0 +1,45 @@ +package com.neusoft.smart.pos.mq; + +/** + * Created by luozukai + */ +public enum EnumMsgMqTopic { + DEFAULT("topic-1","GID_P_1", "默认"), + STOCK("stock","GID_P_2","库存"), + ; + + public static EnumMsgMqTopic getEnum(String code) { + for (EnumMsgMqTopic value : values()) { + if (value.getCode().equals(code)) { + return value; + } + } + return null; + } + + private String code; + private String message; + private String groupId; + + EnumMsgMqTopic(String code, String groupId,String message) { + this.code = code; + this.message = message; + this.groupId = groupId; + } + + public String getCode() { + return code; + } + + public String getMessage() { + return message; + } + + public String getGroupId() { + return groupId; + } + + public void setGroupId(String groupId) { + this.groupId = groupId; + } +} diff --git a/common/src/main/java/com/neusoft/smart/pos/mq/JsonUtil.java b/common/src/main/java/com/neusoft/smart/pos/mq/JsonUtil.java new file mode 100644 index 0000000..76969d6 --- /dev/null +++ b/common/src/main/java/com/neusoft/smart/pos/mq/JsonUtil.java @@ -0,0 +1,128 @@ +package com.neusoft.smart.pos.mq; + + +import com.fasterxml.jackson.annotation.JsonInclude; +import com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.core.JsonParser; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.JsonSerializer; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.SerializerProvider; +import com.fasterxml.jackson.databind.module.SimpleModule; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.io.IOException; +import java.text.SimpleDateFormat; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; + + +public final class JsonUtil { + private static Logger log = LoggerFactory.getLogger(JsonUtil.class); + public static final ObjectMapper mapper = new ObjectMapper(); + + private JsonUtil() { + } + + public static String obj2Json(Object obj) { + if(obj != null) { + try { + return mapper.writeValueAsString(obj); + } catch (JsonProcessingException var2) { + throw new RuntimeException(var2); + } + } else { + return null; + } + } + + public static byte[] obj2Byte(Object obj) { + if(obj != null) { + try { + return mapper.writeValueAsBytes(obj); + } catch (JsonProcessingException var2) { + var2.printStackTrace(); + } + } + + return new byte[0]; + } + + public static Object readValue(String json, Class clazz) { + Object t = null; + + try { + return mapper.readValue(json, clazz); + } catch (Exception var4) { + log.info("解析对象:{}", json); + log.error("json解析错误:{}", var4.getMessage(), var4); + return t; + } + } + +// public static T readValue(String json, TypeReference type) { +// Object object = null; +// +// try { +// object = mapper.readValue(json, type); +// } catch (Exception var4) { +// log.info("解析对象:{}", json); +// log.error("json解析错误:{}", var4.getMessage(), var4); +// } +// +// return object; +// } +// +// public static T readValue(byte[] json, TypeReference type) { +// Object object = null; +// +// try { +// object = mapper.readValue(json, type); +// } catch (Exception var4) { +// log.info("解析对象:{}", json); +// log.error("json解析错误:{}", var4.getMessage(), var4); +// } +// +// return object; +// } +// +// public static Map readValueAsMap(String json) { +// return (Map)readValue(json, new TypeReference>() { +// }); +// } + + static { + mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); + mapper.configure(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS, true); + mapper.configure(DeserializationFeature.ACCEPT_EMPTY_STRING_AS_NULL_OBJECT, true); + mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL); + mapper.setDateFormat(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")); + SimpleModule module = new SimpleModule(); + module.addSerializer(HashMap.class, new JsonUtil.HashMapSerializer()); + mapper.registerModule(module); + } + + public static class HashMapSerializer extends JsonSerializer { + public HashMapSerializer() { + } + + public void serialize(HashMap value, JsonGenerator jgen, SerializerProvider provider) throws IOException { + jgen.writeStartObject(); + Iterator var4 = value.entrySet().iterator(); + + while(var4.hasNext()) { + Object entry = var4.next(); + if(entry instanceof Map.Entry) { + Map.Entry en = (Map.Entry)entry; + if(en.getValue() != null && en.getKey() != null) { + jgen.writeObjectField(en.getKey().toString(), en.getValue()); + } + } + } + + jgen.writeEndObject(); + } + } +} diff --git a/common/src/main/java/com/neusoft/smart/pos/mq/MQConfig.java b/common/src/main/java/com/neusoft/smart/pos/mq/MQConfig.java new file mode 100644 index 0000000..8953eb5 --- /dev/null +++ b/common/src/main/java/com/neusoft/smart/pos/mq/MQConfig.java @@ -0,0 +1,9 @@ +package com.neusoft.smart.pos.mq; + +public class MQConfig { + public static class Impl { + public static final String RABBIT_MQ = "rabbitMQ"; + public static final String ROCKET_MQ = "rocketMQ"; + public static final String ALIYUN_ROCKET_MQ = "aliyunRocketMQ"; + } +} diff --git a/common/src/main/java/com/neusoft/smart/pos/mq/MqProducer.java b/common/src/main/java/com/neusoft/smart/pos/mq/MqProducer.java new file mode 100644 index 0000000..498f0ff --- /dev/null +++ b/common/src/main/java/com/neusoft/smart/pos/mq/MqProducer.java @@ -0,0 +1,7 @@ +package com.neusoft.smart.pos.mq; + +public interface MqProducer { + + public void sendMessage(Object data, String topic, String tags, String keys); + +}