Переглянути джерело

fix aliyunrocketmq

release
xiaohanzi 3 роки тому
джерело
коміт
be3a0197d8
7 змінених файлів з 392 додано та 0 видалено
  1. +10
    -0
      common/pom.xml
  2. +75
    -0
      common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqConsumer.java
  3. +118
    -0
      common/src/main/java/com/neusoft/smart/pos/mq/AliyunRocketMqMessageProducer.java
  4. +45
    -0
      common/src/main/java/com/neusoft/smart/pos/mq/EnumMsgMqTopic.java
  5. +128
    -0
      common/src/main/java/com/neusoft/smart/pos/mq/JsonUtil.java
  6. +9
    -0
      common/src/main/java/com/neusoft/smart/pos/mq/MQConfig.java
  7. +7
    -0
      common/src/main/java/com/neusoft/smart/pos/mq/MqProducer.java

+ 10
- 0
common/pom.xml Переглянути файл

@@ -20,6 +20,16 @@
<artifactId>framework</artifactId>
<version>1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>net.sf.dozer</groupId>
<artifactId>dozer</artifactId>
<version>5.5.1</version>
</dependency>
<dependency>
<groupId>com.aliyun.openservices</groupId>
<artifactId>ons-client</artifactId>
<version>1.8.4.Final</version>
</dependency>
<!-- MySql数据库驱动 -->
<dependency>
<groupId>mysql</groupId>


+ 75
- 0
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));
}
}
}

+ 118
- 0
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<String,Producer> producerMap = null;
@PostConstruct
public void init() {
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(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();
}
}

+ 45
- 0
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;
}
}

+ 128
- 0
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> 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> 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<String, Object> readValueAsMap(String json) {
// return (Map)readValue(json, new TypeReference<HashMap<String, Object>>() {
// });
// }

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<HashMap> {
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();
}
}
}

+ 9
- 0
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";
}
}

+ 7
- 0
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);
}

Завантаження…
Відмінити
Зберегти