【多实例消费MQ实现数据同步的方案】

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092");
props.put("group.id", "sync-data-group");
props.put("enable.auto.commit", "true");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("sync-topic"));

MessageQueueSelector selector = (mqs, msg, arg) -> {
    String orderId = (String) arg;
    int index = orderId.hashCode() % mqs.size();
    return mqs.get(index);
};
producer.send(msg, selector, "ORDER_123");

def is_processed(msg_id):
    key = f"msg:{msg_id}"
    if redis.setnx(key, 1):
        redis.expire(key, 86400)
        return False
    return True

auto.commit.interval.ms=5000
enable.auto.commit=true

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        consumer.commitSync();
    }
}

@Bean
public Queue dlq() {
    return QueueBuilder.durable("sync.DLQ")
           .deadLetterExchange("")
           .deadLetterRoutingKey("sync.retry")
           .build();
}

# Kubernetes Deployment示例
apiVersion: apps/v1
kind: Deployment
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: consumer
        image: sync-service:v1
        env:
        - name: SPRING_PROFILES_ACTIVE
          value: "kafka-consumer"

kafka_consumer_lag{group="sync-data-group"} > 1000