store归因上报

src/main/java/com/lwby/marketing/att/storynovel/StoryNovelActiveConsumer.java

@@ -23,7 +23,7 @@ public class StoryNovelActiveConsumer implements MessageListener<String, String>
     @Autowired
     NovelAttributionMain novelAttributionMain;
 
-    @KafkaListener(topics = {"${store.novel.active.consumer.topic:testStoreNovelActive}"},groupId = "${store.novel.active.consumer.group.id:test_store_dy_active}")
+    @KafkaListener(topics = {"${story.novel.active.consumer.topic:testStoryNovelActive}"},groupId = "${story.novel.active.consumer.group.id:test_story_dy_active}")
     @Override
     public void onMessage(ConsumerRecord<String, String> data) {
 

src/main/java/com/lwby/marketing/controller/AttrController.java

@@ -9,8 +9,10 @@ import com.lwby.marketing.att.storynovel.media.jrtt.JRTTActiveCallback;
 import com.lwby.marketing.att.storynovel.media.jrtt.JRTTPayCallback;
 import com.lwby.marketing.vo.StoryNovelAction;
 import com.lwby.marketing.vo.StoreUserUploadEventBO;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
@@ -24,12 +26,16 @@ import java.util.function.Function;
  */
 
 @RestController
+@Slf4j
 @RequestMapping(value = "/attr")
 public class AttrController {
 
     @Autowired
     NovelAttributionMain nm;
 
+    @Resource
+    private KafkaTemplate<String, String> kafkaTemplate;
+
     @RequestMapping("/testUpload")
     public void testUpload() {
         String msg = "{\"clientInfo\":{\"channel\":214120503,\"clientIp\":\"183.219.7.180\",\"pkv\":1,\"platformGroupId\":412,"
@@ -96,4 +102,54 @@ public class AttrController {
             }
         };
     }
+
+    @RequestMapping("/send")
+    public void testSend() {
+        String msg = "{\"clientInfo\":{\"channel\":214120503,\"clientIp\":\"183.219.7.180\",\"pkv\":1,\"platformGroupId\":412,"
+                + "\"platformId\":412,\"sessionid\":\"undefined\",\"ua\":\"Mozilla/5.0 (Linux; Android 12; OXF-AN00 Build/HUAWEIOXF-AN00;"
+                + " wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/117.0.0.0 Mobile Safari/537.36 aweme.lite/28.9.0 "
+                + "ToutiaoMicroApp/3.14.0 PluginVersion/28909073\",\"version\":\"2.0.0\",\"xClient\":\"version=2.0.0;platformId=412;"
+                + "equipment=android\"},\"createTime\":1709533439693,\"extraData\":{},\"id\":\"93857260-8130-442f-bc92-b549dbf28ef0\","
+                + "\"userId\":875325,\"userUploadEvent\":0,\"appId\":\"ttd3dda5604ce230b401\","
+                + "\"openId\":\"_000HgDjWl-P5-WS9HTQIssNgTnMBRUqDHDu\"}";
+
+        StoreUserUploadEventBO event = JSONObject.parseObject(msg, StoreUserUploadEventBO.class);
+
+        kafkaTemplate.send("testStoryNovelActive",JSON.toJSONString(event)).addCallback(success -> {
+            // 消息发送到的topic
+            String topic = success.getRecordMetadata().topic();
+            // 消息发送到的分区
+            int partition = success.getRecordMetadata().partition();
+            // 消息在分区内的offset
+            long offset = success.getRecordMetadata().offset();
+            log.info("dystory短篇注册发送消息成功: topic={},partition={},offset={}", topic, partition, offset);
+        }, failure -> {
+            log.error("dystory短篇注册发送消息失败:fail={}", JSON.toJSONString(failure));
+        });
+    }
+
+    @RequestMapping("/sendPay")
+    public void testSendPay() {
+        String msg = "{\"clientInfo\":{\"channel\":214120503,\"clientIp\":\"183.219.7.180\",\"pkv\":1,\"platformGroupId\":412,"
+                + "\"platformId\":412,\"sessionid\":\"undefined\",\"ua\":\"Mozilla/5.0 (Linux; Android 12; OXF-AN00 Build/HUAWEIOXF-AN00;"
+                + " wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/117.0.0.0 Mobile Safari/537.36 aweme.lite/28.9.0 "
+                + "ToutiaoMicroApp/3.14.0 PluginVersion/28909073\",\"version\":\"2.0.0\",\"xClient\":\"version=2.0.0;platformId=412;"
+                + "equipment=android\"},\"createTime\":1709533439693,\"extraData\":{},\"id\":\"93857260-8130-442f-bc92-b549dbf28ef0\","
+                + "\"userId\":875325,\"userUploadEvent\":2,\"productId\":\"1\",\"appId\":\"ttd3dda5604ce230b401\","
+                + "\"openId\":\"_000HgDjWl-P5-WS9HTQIssNgTnMBRUqDHDu\"}";
+
+        StoreUserUploadEventBO event = JSONObject.parseObject(msg, StoreUserUploadEventBO.class);
+
+        kafkaTemplate.send("testStoryNovelpay",JSON.toJSONString(event)).addCallback(success -> {
+            // 消息发送到的topic
+            String topic = success.getRecordMetadata().topic();
+            // 消息发送到的分区
+            int partition = success.getRecordMetadata().partition();
+            // 消息在分区内的offset
+            long offset = success.getRecordMetadata().offset();
+            log.info("dystory短篇付费发送消息成功: topic={},partition={},offset={}", topic, partition, offset);
+        }, failure -> {
+            log.error("dystory短篇付费发送消息失败:fail={}", JSON.toJSONString(failure));
+        });
+    }
 }