// 用戶注冊(cè)方法
  public void registerUser(String username, String email, String phoneNumber) {
      // 保存用戶信息（簡(jiǎn)化版）
      userService.add(buildUser(username,email,phoneNumber))
      // 發(fā)送消息
      String registrationMessage = "User " + username + " has registered successfully.";
      // 發(fā)送消息到隊(duì)列
      rabbitTemplate.convertAndSend("registrationQueue", registrationMessage);
  }

@Service
public class NotificationService {

    // 監(jiān)聽(tīng)消息隊(duì)列中的消息并發(fā)送短信/郵件
    @RabbitListener(queues = "registrationQueue")
    public void handleRegistrationNotification(String message) {
        // 這里可以進(jìn)行短信或郵件的發(fā)送操作
        System.out.println("Sending registration notification: " + message);

        // 假設(shè)這里是發(fā)送短信的操作
        sendSms(message);

        // 也可以做其他通知（比如發(fā)郵件等）
        sendEmail(message);
    }
  }

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;

public class PaymentService {

    private DefaultMQProducer producer;

    public PaymentService() throws Exception {
        producer = new DefaultMQProducer("PaymentProducerGroup");
        producer.setNamesrvAddr("localhost:9876");  // RocketMQ NameServer 地址
        producer.start();
    }

    public void processPayment(String orderId, int quantity) throws Exception {
        // 1. 模擬調(diào)用支付接口（例如：支付寶、微信支付等）
        boolean paymentSuccessful = callPayment(orderId, quantity);

        if (paymentSuccessful) {
            // 2. 支付成功后，創(chuàng)建支付消息并發(fā)送到 RocketMQ
            String messageBody = "OrderId: " + orderId + ", Quantity: " + quantity;
            Message message = new Message("paymentTopic", "paymentTag", messageBody.getBytes());
            producer.send(message);    
        }
    }
}

public class InventoryService {

    private DefaultMQPushConsumer consumer;

    public InventoryService() throws Exception {
        consumer = new DefaultMQPushConsumer("InventoryConsumerGroup");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("paymentTopic", "paymentTag");

        // 消息監(jiān)聽(tīng)器
        consumer.registerMessageListener((msgs, context) -> {
            for (MessageExt msg : msgs) {
                String messageBody = new String(msg.getBody());
                // 執(zhí)行扣庫(kù)存操作
                reduceStock(messageBody);
            }
            return null; // 返回消費(fèi)成功
        });

        consumer.start();
        System.out.println("InventoryService started...");
    }
}

@Service
public class OrderService {
 
 @Autowired
 private RocketMQTemplate rocketMQTemplate;
 
 public void createOrder(Order order) {
  // 保存訂單邏輯（省略）
 
  // 計(jì)算延遲時(shí)間（單位：毫秒）
  long delay = order.getTimeout();
 
  // 發(fā)送延遲消息
  rocketMQTemplate.syncSend("orderCancelTopic:delay" + delay,
    MessageBuilder.withPayload(order).build(),
    10000, // 消息發(fā)送超時(shí)時(shí)間（單位：毫秒）
    (int) (delay / 1000) // RocketMQ的延遲級(jí)別是以秒為單位的，因此需要轉(zhuǎn)換為秒
  );
 }
}

@Component
@RocketMQMessageListener(topic = "orderCancelTopic", consumerGroup = "order-cancel-consumer-group")
public class OrderCancelListener implements RocketMQListener<Order> {
 
 @Override
 public void onMessage(Order order) {
  // 取消訂單邏輯
  // 檢查訂單狀態(tài)，如果訂單仍處于未支付狀態(tài)則進(jìn)行取消
  System.out.println("Cancelling order: " + order.getOrderId());
  // （省略實(shí)際的取消訂單邏輯）
 }
}

// 配置和發(fā)送日志到 Kafka 主題 "app-logs"
KafkaProducer<String, String> producer = new KafkaProducer<>(config);
String logMessage = "{\"level\": \"INFO\", \"message\": \"Application started\", \"timestamp\": \"2024-12-29T20:30:59\"}";
producer.send(new ProducerRecord<>("app-logs", "log-key", logMessage));

@Service
public class LogConsumer {
    // 使用 @KafkaListener 注解來(lái)消費(fèi) Kafka 中的日志
    @KafkaListener(topics = "app-logs", groupId = "log-consumer-group")
    public void consumeLog(String logMessage) {
        // 打印或處理收到的日志
        System.out.println("Received log: " + logMessage);
    }
}

// 創(chuàng)建訂單支付成功事件消息
String orderEventData = "{\"orderId\": 12345, \"userId\": 67890, \"amount\": 100.0, \"event\": \"ORDER_PAYMENT_SUCCESS\"}";
Message msg = new Message("order_event_topic", "order_payment_success", orderEventData.getBytes());

// 發(fā)送消息
producer.send(msg);

// 注冊(cè)消息監(jiān)聽(tīng)器
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (Message msg : msgs) {
                String eventData = new String(msg.getBody());
                System.out.println("Inventory system received: " + eventData);
                
                // 處理庫(kù)存減少邏輯
                // 解析消息（假設(shè)是 JSON 格式）
                // updateInventory(eventData);  // 假設(shè)調(diào)用庫(kù)存更新方法
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

// 注冊(cè)消息監(jiān)聽(tīng)器
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (Message msg : msgs) {
                String eventData = new String(msg.getBody());
                System.out.println("Points system received: " + eventData);

                // 處理用戶積分增加邏輯
                // updateUserPoints(eventData);  // 假設(shè)調(diào)用積分更新方法
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

// 注冊(cè)消息監(jiān)聽(tīng)器
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (Message msg : msgs) {
                String eventData = new String(msg.getBody());
                System.out.println("Finance system received: " + eventData);

                // 處理財(cái)務(wù)記錄邏輯
                // recordPaymentTransaction(eventData);  // 假設(shè)調(diào)用財(cái)務(wù)記錄方法
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });