志愿者变动mq

epmet-commons/epmet-commons-rocketmq/src/main/java/com/epmet/commons/rocketmq/constants/ConsomerGroupConstants.java

@@ -88,4 +88,9 @@ public interface ConsomerGroupConstants {
      *  需求完成，如果服务方是区域化党建单位，重新计算这个单位的满意度
      */
     String USER_DEMAND_FINISH_GROUP = "user_demand_finish_group";
+
+    /**
+     * 志愿者变更
+     */
+    String VOLUNTEER_CHANGE_EVENT_LISTENER_GROUP = "volunteer_change_event_listener_group";
 }

epmet-user/epmet-user-server/src/main/java/com/epmet/controller/IcVolunteerPolyController.java

@@ -2,6 +2,7 @@ package com.epmet.controller;
 
 import com.epmet.commons.tools.annotation.LoginUser;
 import com.epmet.commons.tools.aop.NoRepeatSubmit;
+import com.epmet.commons.tools.dto.form.mq.MqBaseFormDTO;
 import com.epmet.commons.tools.page.PageData;
 import com.epmet.commons.tools.security.dto.TokenDto;
 import com.epmet.commons.tools.utils.Result;
@@ -116,4 +117,17 @@ public class IcVolunteerPolyController {
         icVolunteerPolyService.volunteerDataExtraction(customerId);
         return new Result();
     }
+
+    /**
+     * 志愿者变动
+     * @Param form
+     * @Return {@link Result}
+     * @Author zhaoqifeng
+     * @Date 2022/5/19 18:05
+     */
+    @PostMapping("volunteerChanged")
+    public Result volunteerChanged(@RequestBody MqBaseFormDTO form){
+        icVolunteerPolyService.volunteerChanged(form);
+        return new Result();
+    }
 }

epmet-user/epmet-user-server/src/main/java/com/epmet/mq/RocketMQWarnStatsRegister.java

@@ -5,6 +5,7 @@ import com.epmet.commons.rocketmq.constants.TopicConstants;
 import com.epmet.commons.rocketmq.register.MQAbstractRegister;
 import com.epmet.commons.rocketmq.register.MQConsumerProperties;
 import com.epmet.mq.listener.ICWarnStatsEventListener;
+import com.epmet.mq.listener.VolunteerChangeEventListener;
 import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
 import org.springframework.stereotype.Component;
 
@@ -27,5 +28,12 @@ public class RocketMQWarnStatsRegister extends MQAbstractRegister {
                 new ICWarnStatsEventListener());
 
         // ...其他监听器类似
+        // 客户初始化监听器注册
+        register(consumerProperties,
+                ConsomerGroupConstants.VOLUNTEER_CHANGE_EVENT_LISTENER_GROUP,
+                MessageModel.CLUSTERING,
+                TopicConstants.VOLUNTEER,
+                "*",
+                new VolunteerChangeEventListener());
     }
 }

epmet-user/epmet-user-server/src/main/java/com/epmet/mq/listener/VolunteerChangeEventListener.java

@@ -0,0 +1,94 @@
+package com.epmet.mq.listener;
+
+import com.alibaba.fastjson.JSON;
+import com.epmet.commons.rocketmq.constants.MQUserPropertys;
+import com.epmet.commons.tools.distributedlock.DistributedLock;
+import com.epmet.commons.tools.dto.form.mq.MqBaseFormDTO;
+import com.epmet.commons.tools.exception.EpmetException;
+import com.epmet.commons.tools.exception.ExceptionUtils;
+import com.epmet.commons.tools.redis.RedisKeys;
+import com.epmet.commons.tools.redis.RedisUtils;
+import com.epmet.commons.tools.utils.SpringContextUtils;
+import com.epmet.service.IcVolunteerPolyService;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang.StringUtils;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.redisson.api.RLock;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description
+ * @Author zhaoqifeng
+ * @Date 2022/5/19 17:54
+ */
+@Slf4j
+public class VolunteerChangeEventListener implements MessageListenerConcurrently {
+    private RedisUtils redisUtils;
+
+    @Override
+    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+        if (redisUtils == null) {
+            redisUtils = SpringContextUtils.getBean(RedisUtils.class);
+        }
+
+        try {
+            msgs.forEach(msg -> consumeMessage(msg));
+        } catch (Exception e) {
+            log.error(ExceptionUtils.getErrorStackTrace(e));
+            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+    }
+
+    private void consumeMessage(MessageExt messageExt) {
+        // msg即为消息体
+        // tags为SystemMessageType.java中的项，为具体的操作，此处拿到tags，判断是创建还是变更，来做响应的后续操作即可
+        String msg = new String(messageExt.getBody());
+        log.info("msg is {}",msg);
+        String topic = messageExt.getTopic();
+        String tags = messageExt.getTags();
+        String pendingMsgLabel = messageExt.getUserProperty(MQUserPropertys.BLOCKED_MSG_LABEL);
+
+        log.info("【开放数据事件监听器】-志愿者变动-收到消息内容:{},操作:{}", msg, tags);
+        MqBaseFormDTO obj = JSON.parseObject(msg, MqBaseFormDTO.class);
+        log.info("obj is {}",JSON.toJSONString(obj));
+        DistributedLock distributedLock = null;
+        RLock lock = null;
+        try {
+            distributedLock = SpringContextUtils.getBean(DistributedLock.class);
+            lock = distributedLock.getLock(String.format("lock:ic_warn_stats:%s", obj.getCustomerId()),
+                    30L, 30L, TimeUnit.SECONDS);
+            //待执行方法
+            SpringContextUtils.getBean(IcVolunteerPolyService.class).volunteerChanged(obj);
+
+        } catch (EpmetException e) {
+            // 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
+            log.error("【开放数据事件监听器】-志愿者变动MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
+        } catch (Exception e) {
+            // 不是我们自己抛出的异常，可以让MQ重试
+            log.error("【开放数据事件监听器】-志愿者变动MQ失败:".concat(ExceptionUtils.getErrorStackTrace(e)));
+            throw e;
+        } finally {
+            assert distributedLock != null;
+            distributedLock.unLock(lock);
+        }
+
+        if (StringUtils.isNotBlank(pendingMsgLabel)) {
+            try {
+                removePendingMqMsgCache(pendingMsgLabel);
+            } catch (Exception e) {
+                log.error("【开放数据事件监听器】-删除mq阻塞消息缓存失败:{}", ExceptionUtils.getErrorStackTrace(e));
+            }
+        }
+    }
+
+    private void removePendingMqMsgCache(String pendingMsgLabel) {
+        String key = RedisKeys.blockedMqMsgKey(pendingMsgLabel);
+        redisUtils.delete(key);
+    }
+}