import org.apache.kafka.clients.producer.*;
import java.util.Properties;
 
public class PriorityProducer {
    private final Producer<String, String> producer;
    private final String highPriorityTopic;
    private final String mediumPriorityTopic;
    private final String lowPriorityTopic;
    
    public PriorityProducer(String bootstrapServers) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        
        this.producer = new KafkaProducer<>(props);
        this.highPriorityTopic = "high-priority";
        this.mediumPriorityTopic = "medium-priority";
        this.lowPriorityTopic = "low-priority";
    }
    
    public void sendMessage(String key, String message, int priority) {
        String topic;
        
        // 根據(jù)優(yōu)先級(jí)選擇Topic
        switch (priority) {
            case 1: // 高優(yōu)先級(jí)
                topic = highPriorityTopic;
                break;
            case 2: // 中優(yōu)先級(jí)
                topic = mediumPriorityTopic;
                break;
            default: // 低優(yōu)先級(jí)
                topic = lowPriorityTopic;
                break;
        }
        
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, message);
        
        producer.send(record, (metadata, exception) -> {
            if (exception == null) {
                System.out.println("Message sent to " + metadata.topic() + 
                                  " partition " + metadata.partition() + 
                                  " offset " + metadata.offset());
            } else {
                exception.printStackTrace();
            }
        });
    }
    
    public void close() {
        producer.close();
    }
}

import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.*;
 
public class PriorityConsumer {
    private final Consumer<String, String> consumer;
    private final List<String> topics;
    
    public PriorityConsumer(String bootstrapServers, String groupId) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        props.put("group.id", groupId);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "earliest");
        
        this.consumer = new KafkaConsumer<>(props);
        this.topics = Arrays.asList("high-priority", "medium-priority", "low-priority");
    }
    
    public void consumeMessages() {
        // 先訂閱高優(yōu)先級(jí)Topic
        consumer.subscribe(Collections.singletonList("high-priority"));
        
        while (true) {
            // 先嘗試從高優(yōu)先級(jí)Topic獲取消息
            ConsumerRecords<String, String> highPriorityRecords = 
                consumer.poll(Duration.ofMillis(100));
            
            if (!highPriorityRecords.isEmpty()) {
                processRecords(highPriorityRecords);
                continue;
            }
            
            // 如果高優(yōu)先級(jí)沒有消息，嘗試中優(yōu)先級(jí)
            consumer.subscribe(Collections.singletonList("medium-priority"));
            ConsumerRecords<String, String> mediumPriorityRecords = 
                consumer.poll(Duration.ofMillis(100));
            
            if (!mediumPriorityRecords.isEmpty()) {
                processRecords(mediumPriorityRecords);
                consumer.subscribe(Collections.singletonList("high-priority"));
                continue;
            }
            
            // 如果中優(yōu)先級(jí)也沒有消息，處理低優(yōu)先級(jí)
            consumer.subscribe(Collections.singletonList("low-priority"));
            ConsumerRecords<String, String> lowPriorityRecords = 
                consumer.poll(Duration.ofMillis(100));
            
            if (!lowPriorityRecords.isEmpty()) {
                processRecords(lowPriorityRecords);
            }
            
            // 重新訂閱高優(yōu)先級(jí)
            consumer.subscribe(Collections.singletonList("high-priority"));
        }
    }
    
    private void processRecords(ConsumerRecords<String, String> records) {
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("Received message: " + record.value() + 
                              " from topic: " + record.topic() + 
                              " partition: " + record.partition() + 
                              " offset: " + record.offset());
            
            // 處理消息的業(yè)務(wù)邏輯
            processMessage(record.value());
        }
    }
    
    private void processMessage(String message) {
        // 實(shí)際的消息處理邏輯
        System.out.println("Processing message: " + message);
    }
    
    public void close() {
        consumer.close();
    }
}

from kafka import KafkaProducer, KafkaConsumer
import json
import time
 
# 生產(chǎn)者
class PriorityProducer:
    def __init__(self, bootstrap_servers):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8')
        )
        self.topics = {
            1: "high-priority",
            2: "medium-priority",
            3: "low-priority"
        }
    
    def send_message(self, message, priority=3):
        topic = self.topics.get(priority, self.topics[3])
        self.producer.send(topic, message)
        self.producer.flush()
        print(f"Sent message to {topic}: {message}")
    
    def close(self):
        self.producer.close()
 
# 消費(fèi)者
class PriorityConsumer:
    def __init__(self, bootstrap_servers, group_id):
        self.bootstrap_servers = bootstrap_servers
        self.group_id = group_id
        self.topics = ["high-priority", "medium-priority", "low-priority"]
        self.consumers = {}
        
        for topic in self.topics:
            self.consumers[topic] = KafkaConsumer(
                topic,
                bootstrap_servers=bootstrap_servers,
                group_id=f"{group_id}-{topic}",
                value_deserializer=lambda v: json.loads(v.decode('utf-8')),
                auto_offset_reset='earliest'
            )
    
    def consume_with_priority(self):
        while True:
            # 先檢查高優(yōu)先級(jí)消息
            high_priority_messages = list(self.consumers["high-priority"].poll(timeout_ms=100).values())
            if high_priority_messages:
                for message_list in high_priority_messages:
                    for message in message_list:
                        self.process_message(message, "high-priority")
                continue
            
            # 檢查中優(yōu)先級(jí)消息
            medium_priority_messages = list(self.consumers["medium-priority"].poll(timeout_ms=100).values())
            if medium_priority_messages:
                for message_list in medium_priority_messages:
                    for message in message_list:
                        self.process_message(message, "medium-priority")
                continue
            
            # 檢查低優(yōu)先級(jí)消息
            low_priority_messages = list(self.consumers["low-priority"].poll(timeout_ms=100).values())
            if low_priority_messages:
                for message_list in low_priority_messages:
                    for message in message_list:
                        self.process_message(message, "low-priority")
            
            time.sleep(0.01)  # 避免CPU占用過高
    
    def process_message(self, message, topic):
        print(f"Processing {topic} message: {message.value}")
        # 實(shí)際的消息處理邏輯
    
    def close(self):
        for consumer in self.consumers.values():
            consumer.close()