新增Redis消息队列实现

jd-api/pom.xml

@@ -55,6 +55,7 @@
             <groupId>com.alibaba.cloud</groupId>
             <artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
         </dependency>
+
         <dependency>
             <groupId>com.qihang</groupId>
             <artifactId>security</artifactId>

jd-api/src/main/java/com/qihang/jd/JdApi.java

@@ -20,4 +20,11 @@ public class JdApi
         System.out.println( "Hello jd-api!" );
         SpringApplication.run(JdApi.class, args);
     }
+
+    /**
+     * redis消息监听器容器
+     */
+
+
+
 }

jd-api/src/main/java/com/qihang/jd/controller/HomeController.java

@@ -1,12 +1,27 @@
 package com.qihang.jd.controller;
 
+import com.qihang.jd.mq.MqUtils;
+import com.qihang.jd.mq.MqMessage;
+import lombok.AllArgsConstructor;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+@AllArgsConstructor
 @RestController
 public class HomeController {
+    private final MqUtils mqUtils;
+//    private final RedisTemplate redisTemplate;
+//
+//    public void sendMessage(String channel, Object message) {
+//        redisTemplate.convertAndSend(channel, message);
+////        redisTemplate.convertAndSend(channel, message);
+//    }
+
     @GetMapping("/")
     public String home(){
+        MqMessage mqVo = MqMessage.build(1,"52332555");
+
+        mqUtils.sendMessage("channel",mqVo);
         return "{'code':0,'msg':'请通过api访问'}";
     }
 }

jd-api/src/main/java/com/qihang/jd/controller/OrderApiController.java

@@ -3,17 +3,13 @@ package com.qihang.jd.controller;
 import com.jd.open.api.sdk.DefaultJdClient;
 import com.jd.open.api.sdk.JdClient;
 import com.jd.open.api.sdk.request.order.PopOrderEnSearchRequest;
-import com.jd.open.api.sdk.request.refundapply.PopAfsRefundapplyQuerylistRequest;
-import com.jd.open.api.sdk.request.ware.SkuReadSearchSkuListRequest;
-import com.jd.open.api.sdk.request.ware.WareReadSearchWare4ValidRequest;
 import com.jd.open.api.sdk.response.order.PopOrderEnSearchResponse;
-import com.jd.open.api.sdk.response.refundapply.PopAfsRefundapplyQuerylistResponse;
-import com.jd.open.api.sdk.response.ware.SkuReadSearchSkuListResponse;
-import com.jd.open.api.sdk.response.ware.WareReadSearchWare4ValidResponse;
 import com.qihang.common.common.ApiResult;
 import com.qihang.common.enums.HttpStatus;
 import com.qihang.jd.common.ApiCommon;
 import com.qihang.jd.common.PullRequest;
+import com.qihang.jd.mq.MqMessage;
+import com.qihang.jd.mq.MqUtils;
 import lombok.AllArgsConstructor;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -25,8 +21,14 @@ import org.springframework.web.bind.annotation.RestController;
 @AllArgsConstructor
 public class OrderApiController {
     private final ApiCommon apiCommon;
+//    private final RedisCache redisCache;
+    private final MqUtils mqUtils;
+
+
     @RequestMapping(value = "/pull_list", method = RequestMethod.POST)
     public Object pullList(@RequestBody PullRequest params) throws Exception {
+//        Object cacheObject = redisCache.getCacheObject("jdorder");
+
         if (params.getShopId() == null || params.getShopId() <= 0) {
 //            return ApiResul new ApiResult(HttpStatus.PARAMS_ERROR, "参数错误，没有店铺Id");
             return ApiResult.build(HttpStatus.PARAMS_ERROR, "参数错误，没有店铺Id");
@@ -61,7 +63,9 @@ public class OrderApiController {
         PopOrderEnSearchRequest request =new PopOrderEnSearchRequest();
         request.setStartDate("2024-02-06 00:20:35");
         request.setEndDate("2024-03-05 15:20:35");
-        request.setOrderState("WAIT_GOODS_RECEIVE_CONFIRM");
+//        request.setOrderState("WAIT_SELLER_STOCK_OUT,WAIT_GOODS_RECEIVE_CONFIRM,WAIT_SELLER_DELIVERY,PAUSE,FINISHED_L,TRADE_CANCELED,LOCKED,POP_ORDER_PAUSE");
+        request.setOrderState("");
+//        request.setOrderState("ALL");
 //        request.setOptionalFields("orderId,venderId");
 //        request.setSourceId("JOS");
         request.setOptionalFields("venderId,orderId,orderType,payType,orderTotalPrice,orderSellerPrice,orderPayment,freightPrice,sellerDiscount,orderState" +
@@ -73,7 +77,8 @@ public class OrderApiController {
         request.setSortType(1);
         request.setDateType(0);
         PopOrderEnSearchResponse response=client.execute(request);
-
+        MqMessage mqVo = MqMessage.build(1,"52332555000");
+        mqUtils.sendApiMessage(mqVo);
         return response;
     }
 }

jd-api/src/main/java/com/qihang/jd/mq/ApiMessageReceiver.java

@@ -0,0 +1,38 @@
+package com.qihang.jd.mq;
+
+
+import com.alibaba.fastjson2.JSON;
+import lombok.AllArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.serializer.RedisSerializer;
+import org.springframework.stereotype.Component;
+@AllArgsConstructor
+@Component
+public class ApiMessageReceiver implements MessageListener {
+    private static final Logger logger = LoggerFactory.getLogger(ApiMessageReceiver.class);
+
+    private final RedisTemplate redisTemplate;
+
+
+
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        RedisSerializer<String> valueSerializer = redisTemplate.getStringSerializer();
+        String deserialize = valueSerializer.deserialize(message.getBody());
+        //message 和 deserialize 都能拿到msg
+        logger.info("Received <" + message + ">");
+        System.out.println("deserialize = " + deserialize);
+        String messageContent = new String(message.getBody());
+        MqMessage vo = JSON.parseObject(messageContent, MqMessage.class);
+
+        System.out.println(vo.getMqType());
+        if(vo.getMqType() == 1){
+            // 有新订单，插入新订单到shop_order
+            logger.info("有新订单，插入新订单到shop_order");
+        }
+    }
+}

jd-api/src/main/java/com/qihang/jd/mq/MqMessage.java

@@ -0,0 +1,16 @@
+package com.qihang.jd.mq;
+
+import lombok.Data;
+
+@Data
+public class MqMessage {
+    private int mqType;// 消息类型（1：订单消息；2：退款消息）
+    private String keyId;//主键ID
+
+    public static MqMessage build(int mqType , String keyId){
+        MqMessage result = new MqMessage();
+        result.setMqType(mqType);
+        result.setKeyId(keyId);
+        return result;
+    }
+}

jd-api/src/main/java/com/qihang/jd/mq/MqUtils.java

@@ -0,0 +1,18 @@
+package com.qihang.jd.mq;
+
+import lombok.AllArgsConstructor;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+@AllArgsConstructor
+@Component
+public class MqUtils {
+    private final RedisTemplate redisTemplate;
+
+    public void sendMessage(String channel, Object message) {
+        redisTemplate.convertAndSend(channel, message);
+    }
+    public void sendApiMessage(MqMessage message) {
+        redisTemplate.convertAndSend("ApiMessage", message);
+    }
+}

jd-api/src/main/java/com/qihang/jd/mq/SubscriberConfig.java

@@ -0,0 +1,65 @@
+package com.qihang.jd.mq;
+
+import lombok.AllArgsConstructor;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.PatternTopic;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+
+@AllArgsConstructor
+@Configuration
+@AutoConfigureAfter({ApiMessageReceiver.class})
+public class SubscriberConfig {
+    private final RedisTemplate redisTemplate;
+
+    /**
+     * 消息监听适配器，注入接受消息方法，输入方法名字 反射方法
+     *
+     * @param apiMessageReceiver
+     * @return
+     */
+    @Bean
+    public MessageListenerAdapter getMessageListenerAdapter(ApiMessageReceiver apiMessageReceiver) {
+        //当没有实现MessageListener时需要写接收消息的方法名字，实现了就不用写receiveMessage了
+//        return new MessageListenerAdapter(apiMessageReceiver, "receiveMessage");
+        return new MessageListenerAdapter(apiMessageReceiver);
+    }
+
+    /**
+     * 创建消息监听容器
+     *
+     * @param redisConnectionFactory
+     * @param messageListenerAdapter
+     * @return
+     */
+    @Bean
+    public RedisMessageListenerContainer getRedisMessageListenerContainer(RedisConnectionFactory redisConnectionFactory, MessageListenerAdapter messageListenerAdapter) {
+        RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
+        redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
+        redisMessageListenerContainer.addMessageListener(messageListenerAdapter, new PatternTopic("TOPIC_USERNAME"));
+        return redisMessageListenerContainer;
+    }
+
+
+    /**
+     *
+     * @param connectionFactory
+     * @return
+     */
+    @Bean
+    public RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+
+        //订阅频道，通配符*表示任意多个占位符
+
+        // API 消息订阅
+        container.addMessageListener(new ApiMessageReceiver(redisTemplate), new PatternTopic("ApiMessage*"));
+
+        return container;
+    }
+}

jd-api/src/main/java/com/qihang/jd/task/OrderTask.java

@@ -19,7 +19,7 @@ public class OrderTask implements IPollableService {
 
     @Override
     public String getCronExpression() {
-        SysTask task = taskService.getById(1);
+        SysTask task = taskService.getById(2);
 //        return "0/1 * * * * ?";
         return task.getCron();
     }