package com.xiaopeng.automaticPilot.service; import lombok.extern.slf4j.Slf4j; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service; /** * Kafka 消费者服务 */ @Slf4j @Service public class KafkaConsumerService { /** * 消费消息 - 自动驾驶主题 * * @param message 消息内容 */ @KafkaListener(topics = "automatic-pilot-topic", groupId = "xp-automaticPilot-group") public void consumeAutomaticPilotMessage(String message) { try { log.info("收到自动驾驶主题消息: {}", message); // 处理消息逻辑 processMessage(message); } catch (Exception e) { log.error("处理消息失败: {}", message, e); throw e; } } /** * 消费消息 - 车辆状态主题 * * @param message 消息内容 */ @KafkaListener(topics = "vehicle-status-topic", groupId = "xp-automaticPilot-group") public void consumeVehicleStatusMessage(String message) { try { log.info("收到车辆状态主题消息: {}", message); // 处理车辆状态消息逻辑 processVehicleStatus(message); } catch (Exception e) { log.error("处理车辆状态消息失败: {}", message, e); throw e; } } /** * 消费消息 - 导航主题 * * @param message 消息内容 */ @KafkaListener(topics = "navigation-topic", groupId = "xp-automaticPilot-group") public void consumeNavigationMessage(String message) { try { log.info("收到导航主题消息: {}", message); // 处理导航消息逻辑 processNavigation(message); } catch (Exception e) { log.error("处理导航消息失败: {}", message, e); throw e; } } /** * 消费消息 - 订单主题 * * @param message 消息内容 */ @KafkaListener(topics = "order-topic", groupId = "xp-automaticPilot-group") public void consumeOrderMessage(String message) { try { log.info("收到订单主题消息: {}", message); // 处理订单消息逻辑 processOrder(message); } catch (Exception e) { log.error("处理订单消息失败: {}", message, e); throw e; } } /** * 处理消息逻辑(示例) * * @param message 消息内容 */ private void processMessage(String message) { // TODO: 实现具体的消息处理逻辑 log.info("处理自动驾驶消息: {}", message); } /** * 处理车辆状态消息(示例) * * @param message 消息内容 */ private void processVehicleStatus(String message) { // TODO: 实现具体的车辆状态处理逻辑 log.info("处理车辆状态消息: {}", message); } /** * 处理导航消息(示例) * * @param message 消息内容 */ private void processNavigation(String message) { // TODO: 实现具体的导航处理逻辑 log.info("处理导航消息: {}", message); } /** * 处理订单消息(示例) * * @param message 消息内容 */ private void processOrder(String message) { // TODO: 实现具体的订单处理逻辑 log.info("处理订单消息: {}", message); } }