| @@ -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 | |||
| @@ -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: | |||
| @@ -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: | |||
| @@ -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: | |||
| @@ -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<Subscription, MessageListener> subscriptionTable = new HashMap<Subscription, MessageListener>(); | |||
| 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) { | |||
| @@ -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: | |||
| @@ -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: | |||
| @@ -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: | |||
| @@ -26,10 +26,10 @@ | |||
| <version>5.5.1</version> | |||
| </dependency> | |||
| <dependency> | |||
| <groupId>com.aliyun.openservices</groupId> | |||
| <artifactId>ons-client</artifactId> | |||
| <version>1.8.5.Final</version> | |||
| </dependency> | |||
| <groupId>com.aliyun.openservices</groupId> | |||
| <artifactId>ons-client</artifactId> | |||
| <version>1.8.4.Final</version> | |||
| </dependency> | |||
| </dependencies> | |||
| @@ -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<String,Producer> producers = new HashMap<String,Producer>(); | |||
| @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); | |||
| // } | |||
| } | |||
| @@ -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()); | |||