From 6d8ea49c3b482d1f266c8574d69f87d1d0e3fdeb Mon Sep 17 00:00:00 2001 From: xiaohanzi Date: Tue, 9 Jun 2020 23:03:07 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E9=98=BF=E9=87=8C=E4=BA=91ro?= =?UTF-8?q?cketmq?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../src/main/resources/application-dev.yml | 2 +- .../src/main/resources/application-dev.yml | 3 +- .../src/main/resources/application-dev.yml | 3 +- .../src/main/resources/application-dev.yml | 2 +- .../mq/impl/AliyunRocketMqConsumer.java | 107 +++++++++---- .../src/main/resources/application-dev.yml | 4 +- .../src/main/resources/application-dev.yml | 3 +- .../src/main/resources/application-dev.yml | 3 +- mallinkService/pom.xml | 8 +- .../impl/AliyunRocketMqMessageProducer.java | 144 ++++++++++-------- .../service/impl/WxPayOrderServiceImpl.java | 2 +- 11 files changed, 169 insertions(+), 112 deletions(-) diff --git a/mallinkAdmin/src/main/resources/application-dev.yml b/mallinkAdmin/src/main/resources/application-dev.yml index 833935e20..0f74778dd 100644 --- a/mallinkAdmin/src/main/resources/application-dev.yml +++ b/mallinkAdmin/src/main/resources/application-dev.yml @@ -77,7 +77,7 @@ spring: aliyunRocketmq: accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - producerGroupId: GID_P_1 + groupId: GID_P_1 namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 flyway: enabled: false diff --git a/mallinkBApi/src/main/resources/application-dev.yml b/mallinkBApi/src/main/resources/application-dev.yml index 1cd461359..5a8a7f6f1 100644 --- a/mallinkBApi/src/main/resources/application-dev.yml +++ b/mallinkBApi/src/main/resources/application-dev.yml @@ -72,11 +72,10 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - aliyunRocketmq: accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - producerGroupId: GID_P_1 + groupId: GID_P_1 namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: diff --git a/mallinkCApi/src/main/resources/application-dev.yml b/mallinkCApi/src/main/resources/application-dev.yml index 8519f5549..5057809ef 100644 --- a/mallinkCApi/src/main/resources/application-dev.yml +++ b/mallinkCApi/src/main/resources/application-dev.yml @@ -72,11 +72,10 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - aliyunRocketmq: accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - producerGroupId: GID_P_1 + groupId: GID_P_1 namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: diff --git a/mallinkCallback/src/main/resources/application-dev.yml b/mallinkCallback/src/main/resources/application-dev.yml index d15b7dad8..7c49dc492 100644 --- a/mallinkCallback/src/main/resources/application-dev.yml +++ b/mallinkCallback/src/main/resources/application-dev.yml @@ -77,7 +77,7 @@ spring: aliyunRocketmq: accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - producerGroupId: GID_P_1 + groupId: GID_P_1 namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: 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 c4cec5c08..9a6cf8687 100644 --- a/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java +++ b/mallinkMQConsumer/src/main/java/com/iformall/mq/impl/AliyunRocketMqConsumer.java @@ -1,12 +1,16 @@ package com.iformall.mq.impl; +import java.util.HashMap; +import java.util.Map; import java.util.Properties; import javax.annotation.PostConstruct; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Profile; -import org.springframework.stereotype.Service; import com.aliyun.openservices.ons.api.Action; import com.aliyun.openservices.ons.api.ConsumeContext; @@ -15,54 +19,95 @@ 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 com.aliyun.openservices.ons.api.bean.Subscription; +import com.iformall.enums.EnumMsgMqTopic; import com.iformall.mq.MQConfig; import com.iformall.mq.MqBaseConsumer; import lombok.extern.slf4j.Slf4j; @Slf4j -@Service +@Configuration @Profile(MQConfig.Impl.ALIYUN_ROCKET_MQ) public class AliyunRocketMqConsumer extends MqBaseConsumer { - - @Value("${spring.aliyunRocketmq.accessKeyId}") + @Value("${spring.aliyunRocketmq.accessKeyId}") private String accessKeyId; @Value("${spring.aliyunRocketmq.accessKeySecret}") private String accessKeySecret; - - @Value("${spring.aliyunRocketmq.consumerGroupId}") - private String consumerGroupId; + + @Value("${spring.aliyunRocketmq.groupId}") + private String groupId; @Value("${spring.aliyunRocketmq.namesrvAddr}") private String namesrvAddr; - - @PostConstruct - public void init() { - Properties properties = new Properties(); - // 您在控制台创建的 Group ID - properties.put(PropertyKeyConst.GROUP_ID, consumerGroupId); - // 鉴权用 AccessKeyId,在阿里云服务器管理控制台创建 - properties.put(PropertyKeyConst.AccessKey, accessKeyId); - // 鉴权用 AccessKeySecret,在阿里云服务器管理控制台创建 - properties.put(PropertyKeyConst.SecretKey, accessKeySecret); - // 设置 TCP 接入域名,进入控制台的实例详情页面,在页面上方选择实例后,在实例信息中的“获取接入点信息”区域查看 - properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr); - - Consumer consumer = ONSFactory.createConsumer(properties); - consumer.subscribe("topic-1", "*", new MessageListener() { - public Action consume(Message message, ConsumeContext context) { - System.out.println("Receive: " + message); - String msg = new String(message.getBody()); - doMessage(msg); - return Action.CommitMessage; - } + @Bean(initMethod = "start", destroyMethod = "shutdown") + public ConsumerBean buildConsumer() { + ConsumerBean consumerBean = new ConsumerBean(); + //配置文件 + Properties properties = new Properties(); + properties.setProperty(PropertyKeyConst.AccessKey, accessKeyId); + properties.setProperty(PropertyKeyConst.SecretKey, accessKeySecret); + properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, namesrvAddr); + + properties.setProperty(PropertyKeyConst.GROUP_ID, groupId); + //将消费者线程数固定为20个 20为默认值 + properties.setProperty(PropertyKeyConst.ConsumeThreadNums, "20"); + consumerBean.setProperties(properties); + //订阅关系 + Map subscriptionTable = new HashMap(); + Subscription subscription = new Subscription(); + subscription.setTopic(EnumMsgMqTopic.DEFAULT.getCode()); + subscription.setExpression("*"); + subscriptionTable.put(subscription, new MessageListener() { + @Override + public Action consume(Message message, ConsumeContext context) { + log.info("Receive rocket msg: " + message); + try { + doMessage(new String(message.getBody())); + return Action.CommitMessage; + } catch (Exception e) { + //消费失败 + return Action.ReconsumeLater; + } + } + }); - consumer.start(); - System.out.println("Consumer Started"); - } + //订阅多个topic如上面设置 + + consumerBean.setSubscriptionTable(subscriptionTable); + return consumerBean; + } + + + +// @PostConstruct +// public void init() { +// Properties properties = new Properties(); +// // 您在控制台创建的 Group ID +// properties.put(PropertyKeyConst.GROUP_ID, consumerGroupId); +// // 鉴权用 AccessKeyId,在阿里云服务器管理控制台创建 +// properties.put(PropertyKeyConst.AccessKey, accessKeyId); +// // 鉴权用 AccessKeySecret,在阿里云服务器管理控制台创建 +// properties.put(PropertyKeyConst.SecretKey, accessKeySecret); +// // 设置 TCP 接入域名,进入控制台的实例详情页面,在页面上方选择实例后,在实例信息中的“获取接入点信息”区域查看 +// properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr); +// +// Consumer consumer = ONSFactory.createConsumer(properties); +// consumer.subscribe("topic-1", "*", new MessageListener() { +// public Action consume(Message message, ConsumeContext context) { +// System.out.println("Receive: " + message); +// String msg = new String(message.getBody()); +// doMessage(msg); +// return Action.CommitMessage; +// } +// }); +// consumer.start(); +// System.out.println("Consumer Started"); +// } // public void onMessage(String message) { diff --git a/mallinkMQConsumer/src/main/resources/application-dev.yml b/mallinkMQConsumer/src/main/resources/application-dev.yml index cf488b44c..7c2e5b9ee 100644 --- a/mallinkMQConsumer/src/main/resources/application-dev.yml +++ b/mallinkMQConsumer/src/main/resources/application-dev.yml @@ -72,12 +72,10 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - aliyunRocketmq: accessKeyId: "LTAI4G2hFwCBVhxCFbV7nBxf" accessKeySecret: "LTAI4G2hFwCBVhxCFbV7nBxf" - consumerGroupId: "GID_C_1" - producerGroupId: "GID_P_1" + groupId: "GID_P_1" namesrvAddr: "http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080" aws: diff --git a/mallinkPosApi/src/main/resources/application-dev.yml b/mallinkPosApi/src/main/resources/application-dev.yml index 1913382ca..7bbbd9673 100644 --- a/mallinkPosApi/src/main/resources/application-dev.yml +++ b/mallinkPosApi/src/main/resources/application-dev.yml @@ -72,11 +72,10 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - aliyunRocketmq: accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - producerGroupId: GID_P_1 + groupId: GID_P_1 namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: diff --git a/mallinkSchedule/src/main/resources/application-dev.yml b/mallinkSchedule/src/main/resources/application-dev.yml index 536e71e53..8fb701e2b 100644 --- a/mallinkSchedule/src/main/resources/application-dev.yml +++ b/mallinkSchedule/src/main/resources/application-dev.yml @@ -72,11 +72,10 @@ spring: publisher-confirms: true publisher-returns: false virtual-host: / - aliyunRocketmq: accessKeyId: LTAI4G2hFwCBVhxCFbV7nBxf accessKeySecret: LTAI4G2hFwCBVhxCFbV7nBxf - producerGroupId: GID_P_1 + groupId: GID_P_1 namesrvAddr: http://MQ_INST_1796289517488555_Bcqaq2is.cn-beijing.mq-internal.aliyuncs.com:8080 aws: diff --git a/mallinkService/pom.xml b/mallinkService/pom.xml index 058791ccc..e0b95911d 100644 --- a/mallinkService/pom.xml +++ b/mallinkService/pom.xml @@ -26,10 +26,10 @@ 5.5.1 - com.aliyun.openservices - ons-client - 1.8.5.Final - + com.aliyun.openservices + ons-client + 1.8.4.Final + 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 fd3e4539d..9e153655d 100644 --- a/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java +++ b/mallinkService/src/main/java/com/iformall/mq/impl/AliyunRocketMqMessageProducer.java @@ -1,30 +1,31 @@ package com.iformall.mq.impl; -import java.util.HashMap; -import java.util.Map; import java.util.Properties; -import java.util.UUID; +import javax.annotation.PostConstruct; + +import org.junit.Test; +import org.junit.runner.RunWith; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.test.context.SpringBootTest; import org.springframework.context.annotation.Profile; import org.springframework.stereotype.Service; +import org.springframework.test.context.junit4.SpringRunner; 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; +import com.aliyun.openservices.ons.api.bean.OrderProducerBean; +import com.aliyun.openservices.ons.api.exception.ONSClientException; import com.iformall.domain.po.msg.BaseMsg; -import com.iformall.enums.EnumMsgRecordStatus; -import com.iformall.mapper.WxMsgRecordMapper; +import com.iformall.enums.EnumMsgMqTopic; import com.iformall.mq.MQConfig; import com.iformall.mq.MqBaseProducer; -import com.iformall.service.WxMsgRecordService; import com.iformall.utils.JsonUtil; + /** * @Auther: zfy 阿里云rocketmq * @Date: 2019-01-31 @@ -35,77 +36,94 @@ import com.iformall.utils.JsonUtil; public class AliyunRocketMqMessageProducer implements MqBaseProducer { 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.producerGroupId}") - private String producerGroupId; + @Value("${spring.aliyunRocketmq.groupId}") + private String groupId; @Value("${spring.aliyunRocketmq.namesrvAddr}") private String namesrvAddr; - - private static Map producers = new HashMap(); - - @Autowired - private WxMsgRecordMapper wxMsgRecordMapper; - - @Autowired - private WxMsgRecordService wxMsgRecordService; - - private Producer getProducer(String groupId) { - if (producers.containsKey(groupId)) { - return producers.get(groupId); - }else { - Properties properties = new Properties(); - // 您在控制台创建的 Group ID - properties.put(PropertyKeyConst.GROUP_ID, groupId); - // 鉴权用 AccessKeyId,在阿里云服务器管理控制台创建 - properties.put(PropertyKeyConst.AccessKey,accessKeyId); - // 鉴权用 AccessKeySecret,在阿里云服务器管理控制台创建 - properties.put(PropertyKeyConst.SecretKey, accessKeySecret); - // 设置 TCP 接入域名,进入控制台的实例详情页面,在页面上方选择实例后,在实例信息中的“获取接入点信息”区域查看 - properties.put(PropertyKeyConst.NAMESRV_ADDR,namesrvAddr); - - Producer producer = ONSFactory.createProducer(properties); - // 在发送消息前,必须调用 start 方法来启动 Producer,只需调用一次即可 - producer.start(); - producers.put(groupId, producer); - return producer; - } - } + //顺序消息的Producer 已经注册到了spring容器中,后面需要使用时可以直接注入到其它类中 + private static OrderProducerBean orderProducer; + //@Autowired + //private WxMsgRecordMapper wxMsgRecordMapper; + + //@Autowired + //private WxMsgRecordService wxMsgRecordService; + + @PostConstruct + public void init() { + orderProducer = new OrderProducerBean(); + Properties properties = new Properties(); + properties.setProperty(PropertyKeyConst.AccessKey, accessKeyId); + properties.setProperty(PropertyKeyConst.SecretKey, accessKeySecret); + properties.setProperty(PropertyKeyConst.NAMESRV_ADDR, namesrvAddr); + orderProducer.setProperties(properties); + } + @Override public void sendMessage(BaseMsg data, String topic, String tags, String keys) { - try { - data.setUuid(UUID.randomUUID().toString()); - wxMsgRecordService.save(data); - Message msg = new Message( // - // 在控制台创建的 Topic,即该消息所属的 Topic 名称 - topic, - // Message Tag, - // 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在消息队列 RocketMQ 版服务器过滤 - tags, - // Message Body - // 任何二进制形式的数据,消息队列 RocketMQ 版不做任何干预, - // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式 - JsonUtil.obj2Json(data).getBytes()); - // 设置代表消息的业务关键属性,请尽可能全局唯一,以方便您在无法正常收到消息情况下,可通过控制台查询消息并补发 + + String shardingKey = "OrderedKey"; + //循环发送消息 + Message msg = new Message( // + // Message所属的Topic + topic, + // Message Tag 可理解为Gmail中的标签,对消息进行再归类,方便Consumer指定过滤条件在MQ服务器过滤 + tags, + // Message Body 可以是任何二进制形式的数据, MQ不做任何干预 + // 需要Producer与Consumer协商好一致的序列化和反序列化方式 + JsonUtil.obj2Json(data).getBytes()); + // 设置代表消息的业务关键属性,请尽可能全局唯一 + // 以方便您在无法正常收到消息情况下,可通过MQ 控制台查询消息并补发 // 注意:不设置也不会影响消息正常收发 msg.setKey(keys); // 发送消息,只要不抛异常就是成功 - // 打印 Message ID,以便用于消息发送状态查询 - SendResult sendResult = getProducer(producerGroupId).send(msg); - //System.out.println("Send Message success. Message ID is: " + sendResult.getMessageId()); - data.setMsgStatus(EnumMsgRecordStatus.SEND_SUCC.getCode()); - wxMsgRecordMapper.updateByBaseMsg(data); - } catch (Exception e) { - log.error("Message Producer: Send Message Error ", e); - } + try { + //data.setUuid(UUID.randomUUID().toString()); + //wxMsgRecordService.save(data); + SendResult sendResult = orderProducer.send(msg, shardingKey); + assert sendResult != null; + System.out.println(sendResult); + } catch (ONSClientException e) { + System.out.println("发送失败"); + //出现异常意味着发送失败,为了避免消息丢失,建议缓存该消息然后进行重试。 + } + + +// try { +// data.setUuid(UUID.randomUUID().toString()); +// wxMsgRecordService.save(data); +// Message msg = new Message( // +// // 在控制台创建的 Topic,即该消息所属的 Topic 名称 +// topic, +// // Message Tag, +// // 可理解为 Gmail 中的标签,对消息进行再归类,方便 Consumer 指定过滤条件在消息队列 RocketMQ 版服务器过滤 +// tags, +// // Message Body +// // 任何二进制形式的数据,消息队列 RocketMQ 版不做任何干预, +// // 需要 Producer 与 Consumer 协商好一致的序列化和反序列化方式 +// JsonUtil.obj2Json(data).getBytes()); +// // 设置代表消息的业务关键属性,请尽可能全局唯一,以方便您在无法正常收到消息情况下,可通过控制台查询消息并补发 +// // 注意:不设置也不会影响消息正常收发 +// msg.setKey(keys); +// // 发送消息,只要不抛异常就是成功 +// // 打印 Message ID,以便用于消息发送状态查询 +// SendResult sendResult = getProducer(producerGroupId).send(msg); +// //System.out.println("Send Message success. Message ID is: " + sendResult.getMessageId()); +// data.setMsgStatus(EnumMsgRecordStatus.SEND_SUCC.getCode()); +// wxMsgRecordMapper.updateByBaseMsg(data); +// } catch (Exception e) { +// log.error("Message Producer: Send Message Error ", e); +// } } diff --git a/mallinkService/src/main/java/com/iformall/service/impl/WxPayOrderServiceImpl.java b/mallinkService/src/main/java/com/iformall/service/impl/WxPayOrderServiceImpl.java index 159173113..3ac08b67a 100644 --- a/mallinkService/src/main/java/com/iformall/service/impl/WxPayOrderServiceImpl.java +++ b/mallinkService/src/main/java/com/iformall/service/impl/WxPayOrderServiceImpl.java @@ -234,7 +234,7 @@ public class WxPayOrderServiceImpl implements WxPayOrderService { throw new MallinkException(ErrorCode.PAY_ORDER_EXIST); }else if (oldRecord.getPayOrderStatus()==EnumPayStatus.PAY_STATUS_WAIT.getCode()) { logger.error("支付中,等待微信同步状态>>>" + order); - throw new MallinkException(ErrorCode.PAY_ORDER_PAYING.getCode(),"订单支付中,等待微信同步状态"); + throw new MallinkException(ErrorCode.PAY_ORDER_PAYING.getCode(),"订单支付中,等待微信同步状态,防止重复支付!"); }else { //如果是其他状态的,可以重复支付,把状态更新为支付中,不再新增订单 oldRecord.setPayOrderStatus(EnumPayStatus.PAY_STATUS_WAIT.getCode());