• SpringBoot基于AOP实现RocketMQ发送与消费

    一、业务要求

    将http请求参数和返回值部分进行封装，发送给MQ，由于该项目是一个社区项目，需要将用户点赞、评论等信息发送给消费者，使用接口较多，所以采用AOP注解方式进行生产者的消息发送。

    二、AOP几个概念：

    （1）Target：目标类，要被代理的类，例如，UserService；

    （2）JoinPoint（连接点）：所谓的连接点，是指那些被拦截到的方法；

    （3）PointCut（切入点）：被增强的连接点（所谓的增强其实就是添加的新功能）；

    （4）Advice（通知、增强），增强代码；

    （5）Weaving（织入）：是指把增强的advice应用到目标对象target来创建新的代理对象proxy的 过程。

    （6）proxy：代理类；

    （7）Aspect（切面）：是切入点pointcut和通知advice的结合。

    （8）前置通知（@Before）：在我们执行目标方法之前运行；

    （9）后置通知（@After）：在我们执行目标方法结束之后，不管有没有异常；

    （10）返回通知（@AfterReturning）：在我们的目标方法正常返回值后运行；

    （11）异常通知（AfterThrowing）：在我们的目标方法出现异常后运行；

    （12）环绕通知（@Around）：动态代理，需要手动执行jionPoint.process(),其实就是执行我们的目标方法执行之前，相当于前置通知，执行之后就相当于我们的后置通知。

    二、自定义注解@ActivityMqProducer

    任何带有该注解的方法都会将发送mq。

    1. package com.community.annotations;
    2.  
    3. import java.lang.annotation.*;
    4.  
    5. /**
    6.  * @description: 带有该注解的方法将发送活动mq
    7.  * @author kanlina
    8.  * @date 2022/9/1 下午2:10
    9.  * @version 1.0
    10. */
    11. @Target({ElementType.PARAMETER, ElementType.METHOD}) //注解作用域
    12. @Retention(RetentionPolicy.RUNTIME)//注解运行时期
    13. @Documented
    14. public @interface ActivityMqProducer {
    15. }

    三、AOP设置

    1. package com.community.aop;
    2.  
    3. import com.alibaba.fastjson.JSONObject;
    4. import com.community.mq.producer.activityPush.ActivityPushLocal;
    5. import com.marketingCommunityApi.dto.response.CommonModel;
    6. import com.ruubypay.miss.common.response.CommonResponse;
    7. import lombok.extern.slf4j.Slf4j;
    8. import org.aspectj.lang.JoinPoint;
    9. import org.aspectj.lang.annotation.AfterReturning;
    10. import org.aspectj.lang.annotation.Aspect;
    11. import org.aspectj.lang.annotation.Pointcut;
    12. import org.springframework.stereotype.Component;
    13.  
    14. import javax.annotation.Resource;
    15.  
    16. /**
    17.  * @author kanlina
    18.  * @version 1.0
    19.  * @description: 活动mq切入点
    20.  * @date 2022/9/1 下午2:15
    21.  */
    22. @Aspect
    23. @Component
    24. @Slf4j
    25. public class ActivityMqProducerAspect {
    26.     @Resource
    27.     ActivityPushLocal activityPushLocal;
    28.     @Pointcut(value = "@annotation(com.community.annotations.ActivityMqProducer)")
    29.     public void serviceAspect() {
    30.     }
    31.  
    32.     /**
    33.      * 正常返回通知，拦截service层记录用户正常的日志
    34.      */
    35.     @SuppressWarnings({"rawtypes", "unchecked"})
    36.     @AfterReturning(returning = "returnValue", pointcut = "serviceAspect()")
    37.     public void doAfter(JoinPoint joinPoint, Object returnValue) throws Exception {
    38.         log.info("aop处理activityMq开始");
    39.         String name = joinPoint.getSignature().getName();
    40.         System.out.println(name);
    41.         //发送mq
    42.         //获取返回值
    43.         CommonResponse response = (CommonResponse) returnValue;
    44.         CommonModel commonModel = JSONObject.parseObject(JSONObject.parseObject(response.getResData().toString()).getString("CommonModel"), CommonModel.class);
    45.         activityPushLocal.pushMessageLocal(commonModel);
    46.     }
    47. }

    四、MQ配置

    1. 配置消费者和topic 阿里云 消息队列RocketMQ版 创建topic和Group
    2. 生产者配置文件 
      1. "1.0" encoding="UTF-8"?>
      2. <beans xmlns="http://www.springframework.org/schema/beans"
      3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      4.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
      5.  
      6.  
      7.     
      8.     <bean id="activityPushProducerRef" class="com.aliyun.openservices.ons.api.bean.ProducerBean" init-method="start" destroy-method="shutdown">
      9.         
      10.         <property name="properties" >
      11.             <map>
      12.                 
      13.                 <entry key="ProducerId" value="#{messagePushGroup['activityPushProducerId']}"/>
      14.                 
      15.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      16.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      17.             map>
      18.         property>
      19.     bean>
      20.     <bean id="activityPushProducer" class="com.community.mq.producer.PushCenterProducer">
      21.         <property name="topic" value="#{messagePushGroup['activityPushTopicId']}"/>
      22.         <property name="producer" ref="activityPushProducerRef"/>
      23.     bean>
      24.  
      25. beans>
    3.  生产者配置 
      1. package com.community.mq.producer;
      2.  
      3. import com.aliyun.openservices.ons.api.Message;
      4. import com.aliyun.openservices.ons.api.Producer;
      5. import com.aliyun.openservices.ons.api.SendResult;
      6. import org.slf4j.Logger;
      7. import org.slf4j.LoggerFactory;
      8.  
      9. import java.io.UnsupportedEncodingException;
      10.  
      11. /**
      12.  * @description: 消息推送生产者配置
      13.  * @author kanlina
      14.  * @date 2022/8/30 上午10:51
      15.  * @version 1.0
      16. */
      17.  
      18. public class PushCenterProducer {
      19.  
      20.     private Logger logger = LoggerFactory.getLogger(PushCenterProducer.class);
      21.     /**
      22.      * 消息通道：tpoic，生产者和消费者必须保持相同的topic切网络环境能连通才行
      23.      */
      24.     private String topic;
      25.  
      26.     private Producer producer;
      27.  
      28.     public void setTopic(String topic) {
      29.         this.topic = topic;
      30.     }
      31.  
      32.     public void setProducer(Producer producer) {
      33.         this.producer = producer;
      34.     }
      35.  
      36.     /**
      37.      * 消息队列短信队列发送示例
      38.      *  @param message 消息内容
      39.      * @param tag     tag:给消息打的tag，消费者可以消费指定tag的消息，如果消费者配置的 tag接收策略为* ，表示该通道
      40.      *                下的所有tab的消息都要被消费。tag一般用于消息过滤时使用。
      41.      * @param key     给消息设置的一个key，尽量做到全局唯一。
      42.      * @return
      43.      */
      44.     public void pushMessage(String message, String tag, String key) throws UnsupportedEncodingException {
      45.         Message msg = new Message( //
      46.                 // Message所属的Topic
      47.                 topic,
      48.                 // Message Tag 可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在MQ服务器过滤
      49.                 tag,
      50.                 // Message Body 可以是任何二进制形式的数据， MQ不做任何干预
      51.                 // 需要Producer与Consumer协商好一致的序列化和反序列化方式.
      52.                 //短信中心消息通道接收json格式的字符串。（详情查看README.md）严格使用"UTF-8编码"
      53.                 message.getBytes("UTF-8"));
      54.         // 设置代表消息的业务关键属性，请尽可能全局唯一
      55.         // 1 -uuid 2 -uuid+时间戳 3 -推特的snowflake算法（保证唯一）
      56.         // 以方便您在无法正常收到消息情况下，可通过MQ 控制台查询消息并补发
      57.         // 注意：不设置也不会影响消息正常收发
      58.         //此id可以作为业务消费方去重的依据。
      59.         msg.setKey(key);
      60.         // 发送消息，只要不抛异常就是成功 (默认的api就是同步发送消息，只要不抛异常，同步发送消息就是成功)
      61.         try {
      62.             SendResult sendResult = producer.send(msg);
      63.             logger.info("消息管理 推送消息成功 返回参数:{}", sendResult);
      64.         } catch (Exception e) {
      65.             logger.error("消息管理  推送消息异常：", e);
      66.         }
      67.     }
      68. }

    4. 生产者信息发送

      1. package com.community.mq.producer.activityPush;
      2.  
      3. import com.alibaba.fastjson.JSON;
      4. import com.community.mq.producer.PushCenterProducer;
      5. import com.marketingCommunityApi.dto.response.CommonModel;
      6. import com.ruubypay.miss.cmsg.constants.PushType;
      7. import com.ruubypay.miss.cmsg.dto.MqMsgDTO;
      8. import lombok.extern.slf4j.Slf4j;
      9. import org.springframework.scheduling.annotation.Async;
      10. import org.springframework.stereotype.Component;
      11.  
      12. import javax.annotation.Resource;
      13. import java.util.Date;
      14. import java.util.UUID;
      15.  
      16. /**
      17.  * @author kanlina
      18.  * @version 1.0
      19.  * @description: 社区活动推送
      20.  * @date 2022/8/30 上午10:18
      21.  */
      22. @Slf4j
      23. @Component
      24. public class ActivityPushLocal {
      25.  
      26.     @Resource(name = "activityPushProducer")
      27.     private PushCenterProducer activityPushProducer;
      28.  
      29.     /**
      30.      * 发送mq
      31.      *
      32.      * @param smsString
      33.      */
      34.     @Async
      35.     public void smsMessage(String smsString) {
      36.         try {
      37.             activityPushProducer.pushMessage(
      38.                     smsString,
      39.                     "activityPush",
      40.                     UUID.randomUUID().toString().replaceAll("-", ""));
      41.         } catch (Exception e) {
      42.             log.error("调用活动推送异常", e);
      43.         }
      44.     }
      45.  
      46.     /**
      47.      * 消息中心推送消息，只展示在消息页
      48.      */
      49.     @Async
      50.     public void pushMessageLocal(CommonModel model) {
      51.         log.debug("社区消息页，推送消息，推送内容：{}", model);
      52.         smsMessage(JSON.toJSONString(model));
      53.     }
      54. }

    5. 消费者消费

      1. "1.0" encoding="UTF-8"?>
      2. <beans xmlns="http://www.springframework.org/schema/beans"
      3.        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      4.        xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
      5.     <bean id="msgListener" class="com.ruubypay.marketing.mq.message.consumer.MqMessageListener">bean> 
      6.     
      7.     <bean id="transRecordListener" class="com.ruubypay.marketing.mq.message.consumer.MqTransRecordListener"/>
      8.     <bean id="matchedListener" class="com.ruubypay.marketing.mq.message.consumer.MqMatchedListener"/>
      9.     <bean id="matched4CMBListener" class="com.ruubypay.marketing.mq.message.consumer.MqMatched4CMBListener"/>
      10.     <bean id="shopOrderListener" class="com.ruubypay.marketing.mq.message.consumer.MqShopOrderListener"/>
      11.     <bean id="digitalShopOrderListener" class="com.ruubypay.marketing.mq.message.consumer.MqDigitalShopOrderListener"/>
      12.     <bean id="anniversaryListener" class="com.ruubypay.marketing.mq.message.consumer.MqAnniversaryListener"/>
      13.     <bean id="anniversaryOrderListener" class="com.ruubypay.marketing.mq.message.consumer.MqAnniversaryOrderListener"/>
      14.     <bean id="anniversaryBusListener" class="com.ruubypay.marketing.mq.message.consumer.MqAnniversaryBusListener"/>
      15.     <bean id="cmbEquityListener" class="com.ruubypay.marketing.mq.message.consumer.MqCmbEquityListener"/>
      16.     <bean id="groundPushListener" class="com.ruubypay.marketing.mq.message.consumer.MqGroundPushListener"/>
      17.  
      18.     <bean id="iCBCActivityListener" class="com.ruubypay.marketing.mq.message.consumer.MqICBCActivityListener"/>
      19.     <bean id="couponUsingRecordListener" class="com.ruubypay.marketing.mq.message.consumer.MqCouponUsingRecordListener"/>
      20.     <bean id="payFissionListener" class="com.ruubypay.marketing.mq.message.consumer.MqPayFissionListener"/>
      21.     <bean id="payChannelOperateListener" class="com.ruubypay.marketing.mq.message.consumer.MqPayChannelListener"/>
      22.     <bean id="communityActivityPushListener" class="com.ruubypay.marketing.mq.message.consumer.MqCommunityActivityPushListener"/>
      23.  
      24.     
      25.     <bean id="consumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      26.         <property name="properties" > 
      27.             <map>
      28.                 <entry key="GROUP_ID" value="#{messagePushGroup['mq_groupId']}"/>
      29.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      30.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      31.                 <entry key="ConsumeThreadNums" value="50"/>
      32.             map>
      33.         property>
      34.         <property name="subscriptionTable">
      35.             <map>
      36.                 <entry value-ref="msgListener">
      37.                     <key>
      38.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      39.                             <property name="topic" value="#{messagePushGroup['consumerTopic']}"/>
      40.                             <property name="expression" value="*"/>
      41.                             
      42.                         bean>
      43.                     key>
      44.                 entry>
      45.             map>
      46.         property>
      47.     bean>
      48.     <bean id="transRecordConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      49.         <property name="properties" >
      50.             <map>
      51.                 <entry key="GROUP_ID" value="#{messagePushGroup['transRecordGroupId']}"/>
      52.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      53.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      54.             map>
      55.         property>
      56.         <property name="subscriptionTable">
      57.             <map>
      58.                 <entry value-ref="transRecordListener">
      59.                     <key>
      60.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      61.                             <property name="topic" value="#{messagePushGroup['transRecordTopic']}"/>
      62.                             <property name="expression" value="*"/>
      63.                         bean>
      64.                     key>
      65.                 entry>
      66.             map>
      67.         property>
      68.     bean>
      69.     <bean id="shopOrderConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      70.         <property name="properties" >
      71.             <map>
      72.                 <entry key="GROUP_ID" value="#{messagePushGroup['shopOrderGroupId']}"/>
      73.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      74.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      75.             map>
      76.         property>
      77.         <property name="subscriptionTable">
      78.             <map>
      79.                 <entry value-ref="shopOrderListener">
      80.                     <key>
      81.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      82.                             <property name="topic" value="#{messagePushGroup['shopOrderTopic']}"/>
      83.                             <property name="expression" value="*"/>
      84.                         bean>
      85.                     key>
      86.                 entry>
      87.             map>
      88.         property>
      89.     bean>
      90.     <bean id="digitalShopOrderConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      91.         <property name="properties" >
      92.             <map>
      93.                 <entry key="GROUP_ID" value="#{messagePushGroup['digitalShopOrderGroupId']}"/>
      94.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      95.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      96.             map>
      97.         property>
      98.         <property name="subscriptionTable">
      99.             <map>
      100.                 <entry value-ref="digitalShopOrderListener">
      101.                     <key>
      102.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      103.                             <property name="topic" value="#{messagePushGroup['shopOrderTopic']}"/>
      104.                             <property name="expression" value="*"/>
      105.                         bean>
      106.                     key>
      107.                 entry>
      108.             map>
      109.         property>
      110.     bean>
      111.     <bean id="matchedTripConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      112.         <property name="properties" >
      113.             <map>
      114.                 <entry key="GROUP_ID" value="#{messagePushGroup['matchedTripGroupId']}"/>
      115.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      116.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      117.             map>
      118.         property>
      119.         <property name="subscriptionTable">
      120.             <map>
      121.                 <entry value-ref="matchedListener">
      122.                     <key>
      123.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      124.                             <property name="topic" value="#{messagePushGroup['matchedTripTopic']}"/>
      125.                             <property name="expression" value="*"/>
      126.                         bean>
      127.                     key>
      128.                 entry>
      129.             map>
      130.         property>
      131.     bean>
      132.     <bean id="matchedTrip4CMBConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      133.         <property name="properties" >
      134.             <map>
      135.                 <entry key="GROUP_ID" value="#{messagePushGroup['matchedTrip4CMBGroupId']}"/>
      136.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      137.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      138.             map>
      139.         property>
      140.         <property name="subscriptionTable">
      141.             <map>
      142.                 <entry value-ref="matched4CMBListener">
      143.                     <key>
      144.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      145.                             <property name="topic" value="#{messagePushGroup['matchedTripTopic']}"/>
      146.                             <property name="expression" value="*"/>
      147.                         bean>
      148.                     key>
      149.                 entry>
      150.             map>
      151.         property>
      152.     bean>
      153.     <bean id="anniversaryConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      154.         <property name="properties" >
      155.             <map>
      156.                 <entry key="GROUP_ID" value="#{messagePushGroup['anniversary2GroupId']}"/>
      157.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      158.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      159.             map>
      160.         property>
      161.         <property name="subscriptionTable">
      162.             <map>
      163.                 <entry value-ref="anniversaryListener">
      164.                     <key>
      165.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      166.                             <property name="topic" value="#{messagePushGroup['matchedTripTopic']}"/>
      167.                             <property name="expression" value="*"/>
      168.                         bean>
      169.                     key>
      170.                 entry>
      171.             map>
      172.         property>
      173.     bean>
      174.     <bean id="anniversaryOrderConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      175.         <property name="properties" >
      176.             <map>
      177.                 <entry key="GROUP_ID" value="#{messagePushGroup['anniversary2OrderGroupId']}"/>
      178.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      179.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      180.             map>
      181.         property>
      182.         <property name="subscriptionTable">
      183.             <map>
      184.                 <entry value-ref="anniversaryOrderListener">
      185.                     <key>
      186.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      187.                             <property name="topic" value="#{messagePushGroup['shopOrderTopic']}"/>
      188.                             <property name="expression" value="*"/>
      189.                         bean>
      190.                     key>
      191.                 entry>
      192.             map>
      193.         property>
      194.     bean>
      195.     <bean id="anniversaryBusConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      196.         <property name="properties" >
      197.             <map>
      198.                 <entry key="GROUP_ID" value="#{messagePushGroup['anniversary2BusGroupId']}"/>
      199.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      200.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      201.             map>
      202.         property>
      203.         <property name="subscriptionTable">
      204.             <map>
      205.                 <entry value-ref="anniversaryBusListener">
      206.                     <key>
      207.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      208.                             <property name="topic" value="#{messagePushGroup['transRecordTopic']}"/>
      209.                             <property name="expression" value="*"/>
      210.                         bean>
      211.                     key>
      212.                 entry>
      213.             map>
      214.         property>
      215.     bean>
      216.     <bean id="cmbEquityConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      217.         <property name="properties" >
      218.             <map>
      219.                 <entry key="GROUP_ID" value="#{messagePushGroup['cmbEquityGroupId']}"/>
      220.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      221.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      222.             map>
      223.         property>
      224.         <property name="subscriptionTable">
      225.             <map>
      226.                 <entry value-ref="cmbEquityListener">
      227.                     <key>
      228.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      229.                             <property name="topic" value="#{messagePushGroup['cmbEquityTopic']}"/>
      230.                             <property name="expression" value="*"/>
      231.                         bean>
      232.                     key>
      233.                 entry>
      234.             map>
      235.         property>
      236.     bean>
      237.  
      238.     <bean id="groundPushConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      239.         <property name="properties" >
      240.             <map>
      241.                 <entry key="GROUP_ID" value="#{messagePushGroup['groundPushGroupId']}"/>
      242.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      243.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      244.             map>
      245.         property>
      246.         <property name="subscriptionTable">
      247.             <map>
      248.                 <entry value-ref="groundPushListener">
      249.                     <key>
      250.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      251.                             <property name="topic" value="#{messagePushGroup['groundPushTopic']}"/>
      252.                             <property name="expression" value="*"/>
      253.                         bean>
      254.                     key>
      255.                 entry>
      256.             map>
      257.         property>
      258.     bean>
      259.  
      260.  
      262.         <bean id="iCBCActivityConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      263.             <property name="properties" >
      264.                 <map>
      265.                     <entry key="GROUP_ID" value="#{messagePushGroup['iCBCActivityGroupId']}"/>
      266.                     <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      267.                     <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      268.                 map>
      269.             property>
      270.             <property name="subscriptionTable">
      271.                 <map>
      272.                     <entry value-ref="iCBCActivityListener">
      273.                         <key>
      274.                             <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      275.                                 <property name="topic" value="#{messagePushGroup['iCBCActivityTopic']}"/>
      276.                                 <property name="expression" value="*"/>
      277.                             bean>
      278.                         key>
      279.                     entry>
      280.                 map>
      281.             property>
      282.         bean>
      283.  
      284.     
      285.     <bean id="couponUsingRecordConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      286.         <property name="properties" >
      287.             <map>
      288.                 <entry key="GROUP_ID" value="#{messagePushGroup['couponUsingRecordGroupId']}"/>
      289.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      290.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      291.             map>
      292.         property>
      293.         <property name="subscriptionTable">
      294.             <map>
      295.                 <entry value-ref="couponUsingRecordListener">
      296.                     <key>
      297.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      298.                             <property name="topic" value="#{messagePushGroup['couponUsingRecordTopic']}"/>
      299.                             <property name="expression" value="*"/>
      300.                         bean>
      301.                     key>
      302.                 entry>
      303.             map>
      304.         property>
      305.     bean>
      306.  
      307.     
      308.     <bean id="payFissionConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      309.         <property name="properties" >
      310.             <map>
      311.                 <entry key="GROUP_ID" value="#{messagePushGroup['payFissionGroupId']}"/>
      312.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      313.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      314.             map>
      315.         property>
      316.         <property name="subscriptionTable">
      317.             <map>
      318.                 <entry value-ref="payFissionListener">
      319.                     <key>
      320.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      321.                             <property name="topic" value="#{messagePushGroup['groundPushTopic']}"/>
      322.                             <property name="expression" value="*"/>
      323.                         bean>
      324.                     key>
      325.                 entry>
      326.             map>
      327.         property>
      328.     bean>
      329.  
      330.     <bean id="payChannelOperateConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      331.         <property name="properties" >
      332.             <map>
      333.                 <entry key="GROUP_ID" value="#{messagePushGroup['payChannelOperateGroupId']}"/>
      334.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      335.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      336.             map>
      337.         property>
      338.         <property name="subscriptionTable">
      339.             <map>
      340.                 <entry value-ref="payChannelOperateListener">
      341.                     <key>
      342.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      343.                             <property name="topic" value="#{messagePushGroup['payChannelOperateTopic']}"/>
      344.                             <property name="expression" value="*"/>
      345.                         bean>
      346.                     key>
      347.                 entry>
      348.             map>
      349.         property>
      350.     bean>
      351.  
      352.  
      353.     <bean id="communityActivityPushConsumer" class="com.aliyun.openservices.ons.api.bean.ConsumerBean" init-method="start" destroy-method="shutdown">
      354.         <property name="properties" >
      355.             <map>
      356.                 <entry key="GROUP_ID" value="#{messagePushGroup['communityActivityPushGroupId']}"/>
      357.                 <entry key="AccessKey" value="#{messagePushGroup['accessKey']}"/>
      358.                 <entry key="SecretKey" value="#{messagePushGroup['secretKey']}"/>
      359.             map>
      360.         property>
      361.         <property name="subscriptionTable">
      362.             <map>
      363.                 <entry value-ref="communityActivityPushListener">
      364.                     <key>
      365.                         <bean class="com.aliyun.openservices.ons.api.bean.Subscription">
      366.                             <property name="topic" value="#{messagePushGroup['communityActivityPushTopicId']}"/>
      367.                             <property name="expression" value="*"/>
      368.                         bean>
      369.                     key>
      370.                 entry>
      371.             map>
      372.         property>
      373.     bean>
      374.  
      375. beans>

    6. 消费者业务监听

    1. package com.ruubypay.marketing.mq.message.consumer;
    2.  
    3. import com.alibaba.fastjson.JSONObject;
    4. import com.aliyun.openservices.ons.api.Action;
    5. import com.aliyun.openservices.ons.api.ConsumeContext;
    6. import com.aliyun.openservices.ons.api.Message;
    7. import com.aliyun.openservices.ons.api.MessageListener;
    8. import com.marketingCommunityApi.dto.response.CommonModel;
    9. import com.ruubypay.log.annotation.LogMarker;
    10. import com.ruubypay.marketing.mq.model.PayChannelInfo;
    11. import com.ruubypay.marketing.mq.service.ActivityCommunityPushService;
    12. import com.ruubypay.marketing.mq.service.ActivityPayChannelService;
    13. import lombok.extern.slf4j.Slf4j;
    14. import org.springframework.beans.factory.annotation.Autowired;
    15.  
    16. /**
    17.  * @Author KanLina
    18.  * @Description 社区操作通知
    19.  * @Date 2022/9/01 下午2:50
    20.  **/
    21. @Slf4j
    22. public class MqCommunityActivityPushListener implements MessageListener {
    23.  
    24.     @Autowired
    25.     private ActivityCommunityPushService activityCommunityPushService;
    26.  
    27.     @LogMarker(businessDescription = "社区活动操作通知")
    28.     @Override
    29.     public Action consume(Message message, ConsumeContext context) {
    30.         log.debug("收到消息,Topic:{},MsgID:{},开始执行[{}]", message.getTopic(), message.getMsgID(), message.getReconsumeTimes());
    31.         try {
    32.             CommonModel request = JSONObject.parseObject(message.getBody(), CommonModel.class);
    33.             activityCommunityPushService.dealRequest(request);
    34.         } catch (Exception e) {
    35.             log.error("执行异常", e);
    36.             return Action.ReconsumeLater;
    37.         }
    38.         log.debug("收到消息,Topic:{},MsgID:{},处理成功[{}]",message.getTopic(), message.getMsgID(), message.getReconsumeTimes());
    39.         return Action.CommitMessage;
    40.     }
    41. }

     

  • 相关阅读:
    Vector-常用CAN工具 - CANoe入门到精通_02
    PDF流前端如何接收：深度解析与实用策略
    MATLAB冒号符号
    SpringBoot-自定义Starter
    Redis的安装和使用
    阿里发布大模型发布图结构长文本处理智能体，超越GPT-4-128k
    练[MRCTF2020]套娃
    使用patch-package保存node_modules包修改
    深入了解汽车级功率MOSFET NVMFS2D3P04M8LT1G P沟道数据表
    【强化学习】Offline RL
  • 原文地址：https://blog.csdn.net/kk_lina/article/details/126642060