添加积分消息队列接口

xx-activity/xx-activity-server/src/main/java/com/xxfc/platform/activity/biz/IntegralUserRecordBiz.java

@@ -61,8 +61,9 @@ public class IntegralUserRecordBiz extends BaseBiz<IntegralUserRecordMapper, Int
                 return ObjectRestResponse.createFailedResult(1202, "积分规则不存在");
             }
             Integer point = 0;
-            Integer amount = Integer.parseInt(new BigDecimal(integralUserRecord.getAmount()).divide(new BigDecimal("100"), 0, BigDecimal.ROUND_DOWN).toString());
+
             if(ruleObjectRestResponse.getData().getPoint() == 0) {//没有基础分需要计算分数
+                Integer amount = Integer.parseInt(new BigDecimal(integralUserRecord.getAmount()).divide(new BigDecimal("100"), 0, BigDecimal.ROUND_DOWN).toString());
                 JSONObject jsonObject = JSONObject.parseObject(ruleObjectRestResponse.getData().getOtherRule());
                 log.info("查询的其他规则json信息：jsonObject = {}", jsonObject);
                 if(jsonObject ==  null) {

xx-activity/xx-activity-server/src/main/java/com/xxfc/platform/activity/handler/IntegralMQHandler.java

@@ -2,14 +2,23 @@ package com.xxfc.platform.activity.handler;
 
 
 import com.alibaba.fastjson.JSONObject;
+import com.rabbitmq.client.Channel;
 import com.xxfc.platform.activity.biz.IntegralUserRecordBiz;
 import com.xxfc.platform.activity.vo.IntegralUserRecordDto;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.amqp.core.Message;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.amqp.support.AmqpHeaders;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.messaging.handler.annotation.Headers;
 import org.springframework.stereotype.Component;
 
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 @Component
 @Slf4j
 public class IntegralMQHandler {
@@ -18,17 +27,36 @@ public class IntegralMQHandler {
     IntegralUserRecordBiz integralUserRecordBiz;
 
     @RabbitListener(queues = "integral_queue")
-    public void integralHandler(String json) {
-        log.info("接收到的消息：json = {}", json);
+    public void integralHandler(Message message, @Headers Map<String, Object> headers, Channel channel) {
         try{
-            if(StringUtils.isNotBlank(json)) {
-                IntegralUserRecordDto integralUserRecordDto = JSONObject.parseObject(json, IntegralUserRecordDto.class);
-                integralUserRecordBiz.add(integralUserRecordDto);
-            }
+            String messageId = message.getMessageProperties().getMessageId();
+            String msg = new String(message.getBody(), "UTF-8");
+            log.info("接收到的消息：msg = {}, 消息ID是：messageId = {} ", msg, messageId);
+            ExecutorService executorService = Executors.newCachedThreadPool();
+            executorService.execute(new Runnable() {
+                @Override
+                public void run() {
+                    if(StringUtils.isNotBlank(msg)) {
+                        IntegralUserRecordDto integralUserRecordDto = JSONObject.parseObject(msg, IntegralUserRecordDto.class);
+                        integralUserRecordBiz.add(integralUserRecordDto);
+                    }
+                }
+            });
+            executorService.shutdown();
+            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
+            // 手动签收
+            channel.basicAck(deliveryTag, false);
         }catch (Exception e){
             log.info("接收到的消息失败");
+            try{
+                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
+            }catch (IOException i){
+                i.printStackTrace();
+            }
             e.printStackTrace();
         }
 
     }
+
+
 }

xx-universal/xx-universal-server/src/main/java/com/xxfc/platform/universal/biz/MQServiceBiZ.java

 package com.xxfc.platform.universal.biz;
 
 import com.github.wxiaoqi.security.common.msg.ObjectRestResponse;
-import com.xxfc.platform.universal.entity.Dictionary;
 import lombok.AllArgsConstructor;
-import org.apache.poi.ss.formula.functions.T;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageBuilder;
+import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.stereotype.Service;
 
+import java.util.UUID;
+
 @Service
 @AllArgsConstructor
 public class MQServiceBiZ {
@@ -14,7 +17,10 @@ public class MQServiceBiZ {
     private RabbitTemplate rabbitTemplate;
 
     public ObjectRestResponse sendMessage(String exchange, String routKey, String json) {
-        rabbitTemplate.convertAndSend(exchange, routKey, json);
+        Message message = MessageBuilder.withBody(json.getBytes())
+                .setContentType(MessageProperties.CONTENT_TYPE_JSON).setContentEncoding("utf-8")
+                .setMessageId(UUID.randomUUID() + "").build();
+        rabbitTemplate.convertAndSend(exchange, routKey, message);
         return ObjectRestResponse.succ();
     }
 }