架构关系-建立父类-TopicMessageListener

src/main/java/cn/meteor/beyondclouds/core/listener/TopicMessageListener.java

+package cn.meteor.beyondclouds.core.listener;
+
+import cn.meteor.beyondclouds.modules.queue.message.DataItemUpdateMessage;
+import cn.meteor.beyondclouds.modules.queue.message.SearchItemUpdateType;
+import cn.meteor.beyondclouds.modules.search.enums.SearchItemType;
+import cn.meteor.beyondclouds.util.JsonUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.kafka.annotation.KafkaListener;
+
+import java.io.Serializable;
+import java.util.Optional;
+
+/**
+ * @author meteor
+ */
+@Slf4j
+public class TopicMessageListener {
+
+    @KafkaListener(topics = "${beyondclouds.kafka.topics.search-item-update}")
+    public final void itemUpdate(ConsumerRecord<?, String> record) {
+        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
+        if (kafkaMessage.isPresent()) {
+            DataItemUpdateMessage itemUpdateMessage;
+            try {
+                itemUpdateMessage = JsonUtils.toBean(kafkaMessage.get(), DataItemUpdateMessage.class);
+                log.debug("接收到kafka消息：{}", itemUpdateMessage.toString());
+
+                // 调用消息处理函数
+                onItemUpdate(itemUpdateMessage);
+
+            } catch (Exception e) {
+                e.printStackTrace();
+                log.error("DataItemUpdateMessage consume failed:{}", e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 数据更新事件
+     * @param dataItemUpdateMessage
+     */
+    public void onItemUpdate(DataItemUpdateMessage dataItemUpdateMessage) throws Exception {
+
+    }
+
+}

src/main/java/cn/meteor/beyondclouds/modules/search/listener/ISearchItemUpdateListener.java

-package cn.meteor.beyondclouds.modules.search.listener;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-
-/**
- * @author meteor
- */
-public interface ISearchItemUpdateListener {
-
-    /**
-     * 搜索条目更新
-     * @param record
-     */
-    void onSearchItemUpdate(ConsumerRecord<?, String> record);
-
-}

src/main/java/cn/meteor/beyondclouds/modules/search/listener/impl/SearchItemUpdateListener.java → src/main/java/cn/meteor/beyondclouds/modules/search/listener/SearchItemUpdateListener.java

-package cn.meteor.beyondclouds.modules.search.listener.impl;
+package cn.meteor.beyondclouds.modules.search.listener;
 
+import cn.meteor.beyondclouds.core.listener.TopicMessageListener;
 import cn.meteor.beyondclouds.modules.queue.message.DataItemUpdateMessage;
 import cn.meteor.beyondclouds.modules.queue.message.SearchItemUpdateType;
 import cn.meteor.beyondclouds.modules.search.enums.SearchItemType;
-import cn.meteor.beyondclouds.modules.search.listener.ISearchItemUpdateListener;
 import cn.meteor.beyondclouds.modules.search.service.ISearchService;
-import cn.meteor.beyondclouds.util.JsonUtils;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
 import java.io.Serializable;
-import java.util.Optional;
 
 /**
  * @author meteor
  */
 @Component
 @Slf4j
-public class SearchItemUpdateListener implements ISearchItemUpdateListener {
+public class SearchItemUpdateListener extends TopicMessageListener {
 
     private ISearchService searchService;
 
@@ -30,40 +26,26 @@ public class SearchItemUpdateListener implements ISearchItemUpdateListener {
     }
 
     @Override
-    @KafkaListener(topics = "${beyondclouds.kafka.topics.search-item-update}")
-    public void onSearchItemUpdate(ConsumerRecord<?, String> record) {
-        Optional<String> kafkaMessage = Optional.ofNullable(record.value());
-        if (kafkaMessage.isPresent()) {
-            DataItemUpdateMessage itemUpdateMessage;
-
-            try {
-                itemUpdateMessage = JsonUtils.toBean(kafkaMessage.get(), DataItemUpdateMessage.class);
-                log.debug("接收到kafka消息：{}", itemUpdateMessage.toString());
-
-                // 处理搜索条目更新
-                SearchItemUpdateType updateType = itemUpdateMessage.getUpdateType();
-                Serializable itemId = itemUpdateMessage.getItemId();
-                SearchItemType searchItemType = itemUpdateMessage.getItemType();
-
-                // 根据不同的更新类型调用对应的方法
-                switch (updateType) {
-                    case ADD:
-                        searchService.saveSearchItem(searchItemType, String.valueOf(itemId) );
-                        break;
-                    case DELETE:
-                        searchService.deleteSearchItem(searchItemType, String.valueOf(itemId));
-                        break;
-                    case UPDATE:
-                        searchService.updateSearchItem(searchItemType, String.valueOf(itemId));
-                        break;
-                    default:
-                        break;
-                }
-            } catch (Exception e) {
-                e.printStackTrace();
-                log.error("searchItemUpdateMessage covert failed:{}", e.getMessage());
-            }
+    public void onItemUpdate(DataItemUpdateMessage dataItemUpdateMessage) throws Exception {
+
+        // 处理搜索条目更新
+        SearchItemUpdateType updateType = dataItemUpdateMessage.getUpdateType();
+        Serializable itemId = dataItemUpdateMessage.getItemId();
+        SearchItemType searchItemType = dataItemUpdateMessage.getItemType();
+
+        // 根据不同的更新类型调用对应的方法
+        switch (updateType) {
+            case ADD:
+                searchService.saveSearchItem(searchItemType, String.valueOf(itemId) );
+                break;
+            case DELETE:
+                searchService.deleteSearchItem(searchItemType, String.valueOf(itemId));
+                break;
+            case UPDATE:
+                searchService.updateSearchItem(searchItemType, String.valueOf(itemId));
+                break;
+            default:
+                break;
         }
-
     }
 }