import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.CompletableFuture;


@Component
@RocketMQMessageListener(topic = "orderTopic", consumerGroup = "orderGroup")
publicclass OrderConsumer implements RocketMQListener<String> {
    private ExecutorService executor = Executors.newFixedThreadPool(4); // 根據CPU核數優(yōu)化線程數
    
    @Override
    public void onMessage(String message) {
        // 傳統慢速方法：每條消息都同步查數據庫（耗時IO）
        // processSingleMessage(message);  // 替換為批處理優(yōu)化
        
        executor.execute(() -> batchProcessMessages(message));
    }


    // 優(yōu)化后的批處理方法：批量處理多條消息
    public void batchProcessMessages(String message) {
        CompletableFuture.runAsync(() -> {
            try {
                // 模擬復雜邏輯：先聚合消息（如緩存到內存隊列）
                List<String> messages = loadBatchFromMemory(); // 從緩存批量取消息
                
                if (messages.size() >= 100) { // 批處理100條一次
                    // 批量數據庫更新（減少IO次數）
                    for (String msg : messages) {
                        updateDatabase(msg); // 異步或并行執(zhí)行
                    }
                    messages.clear();
                }
                // 消息處理完成后，模擬異步日志
                log.info("Processed message in batch: " + message);
            } catch (Exception e) {
                log.error("Error processing batch", e);
            }
        }, executor);
    }


    private void updateDatabase(String msg) {
        // 假設數據庫更新操作（異步優(yōu)化可改用JDBC批處理）
        System.out.println("數據庫更新：" + msg);
    }
}

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.message.MessageExt;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue;


publicclass PriorityConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("priorityGroup");
        consumer.setNamesrvAddr("localhost:9876"); // MQ服務器地址
        consumer.subscribe("orderTopic", "*");  // 訂閱所有消息


        // 創(chuàng)建優(yōu)先級隊列（PriorityBlockingQueue）
        PriorityBlockingQueue<MessageExt> priorityQueue = new PriorityBlockingQueue<>(1000, 
            (m1, m2) -> m1.getPriority() - m2.getPriority()); // 基于消息優(yōu)先級排序
        
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    int priority = Integer.parseInt(msg.getProperty("priority")); // 獲取消息優(yōu)先級屬性
                    priorityQueue.add(msg);  // 入隊到優(yōu)先級隊列
                }
                
                // 優(yōu)先處理高優(yōu)先級消息
                while (!priorityQueue.isEmpty()) {
                    MessageExt highPriorityMsg = priorityQueue.poll(); // 取最高優(yōu)先級
                    processMessage(highPriorityMsg); // 消費邏輯
                    if (highPriorityMsg.getPriority() > 5) { // 例如，設置支付消息優(yōu)先級高
                        // 加速處理，并跳過普通消息
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }


    private void processMessage(MessageExt msg) {
        String body = new String(msg.getBody());
        System.out.println("處理消息: " + body + ", 優(yōu)先級: " + msg.getProperty("priority"));
    }
}

import com.google.common.util.concurrent.RateLimiter;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import java.util.concurrent.TimeUnit;


publicclass ThrottledProducer {
    privatestatic RateLimiter rateLimiter = RateLimiter.create(100.0); // 限流100條/秒


    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("throttleGroup");
        producer.start();
        
        for (int i = 0; i < 1000000; i++) {  // 準備發(fā)送100萬條
            boolean acquired = rateLimiter.tryAcquire(1, 100, TimeUnit.MILLISECONDS); // 嘗試獲取令牌
            if (acquired) {
                Message msg = new Message("orderTopic", "tagA", ("Message " + i).getBytes());
                producer.send(msg); // 安全發(fā)送
            } else {
                // 隊列積壓時暫停生產（模擬MQ監(jiān)控回調）
                if (checkQueueBacklog() > 50000) { // 自定義函數檢查MQ當前積壓數
                    Thread.sleep(100); // 暫停100ms減發(fā)送頻率
                }
            }
        }
        producer.shutdown();
    }


    // 自定義函數：監(jiān)控MQ積壓（偽代碼）
    private static int checkQueueBacklog() {
        // 通過MQ API獲取當前隊列消息數
        return100000; // 返回值模擬實際場景
    }
}

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Map;


@Configuration
publicclass DLQConfig {
    // 定義主隊列和綁定
    @Bean
    public Queue mainQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-dead-letter-exchange", "dlqExchange"); // 死信路由到指定Exchange
        args.put("x-dead-letter-routing-key", "dlqKey");   // 死信Routing Key
        returnnew Queue("mainQueue", true, false, false, args);
    }


    @Bean
    public Binding binding() {
        return BindingBuilder.bind(mainQueue()).to(exchange()).with("mainKey");
    }


    // 死信隊列和綁定
    @Bean
    public Queue dlqQueue() {
        returnnew Queue("deadLetterQueue");
    }


    @Bean
    public DirectExchange dlqExchange() {
        returnnew DirectExchange("dlqExchange");
    }


    @Bean
    public Binding dlqBinding() {
        return BindingBuilder.bind(dlqQueue()).to(dlqExchange()).with("dlqKey");
    }
}

import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import javax.annotation.PostConstruct;
import java.util.function.Supplier;


@SpringBootApplication
publicclass MQMonitor {
    public static void main(String[] args) {
        SpringApplication.run(MQMonitor.class, args);
    }


    @PostConstruct
    public void setupMonitor(MeterRegistry registry) {
        Supplier<Number> backlogProvider = () -> {
            // 調用MQ admin API獲取當前積壓數（e.g., RocketMQ broker stats）
            return100000; // 返回值模擬實時數據
        };
        
        Gauge.builder("mq.backlog.count", backlogProvider)
            .description("MQ消息積壓量")
            .register(registry);
        
        // 自動化腳本觸發(fā)器（定時檢查）
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            int backlog = backlogProvider.get().intValue();
            if (backlog > 50000) {  // 告警閾值
                System.out.println("MQ積壓量高！啟動自動化修復...");
                autoScaleConsumers(backlog);  // 自動擴容消費者
            }
        }));
    }


    private void autoScaleConsumers(int backlog) {
        // 調用K8s或Ansible API動態(tài)增加消費者實例
        // e.g., 每增加10000條積壓，啟動一個POD
        System.out.println("Auto scaling: added " + (backlog / 10000) + " consumers");
    }
}