IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Consumer案例

wen java案例 94

本文目录导读：

Consumer案例

  1. 基础消费者案例
  2. 手动提交偏移量
  3. 精准一次性消费模式
  4. 多线程消费者案例
  5. 指定分区消费
  6. 消费者组动态分配监控
  7. 实战：带重试和死信队列的消费者
  8. 性能优化案例
  9. 关键配置说明

我来为你详细解析Kafka Consumer的完整案例，包括基础消费、高级特性和实战场景。

基础消费者案例

最简单的消费者

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class SimpleConsumer {
    public static void main(String[] args) {
        // 1. 配置消费者
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 2. 创建消费者
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 3. 订阅主题
        consumer.subscribe(Collections.singletonList("test-topic"));
        try {
            // 4. 消费消息
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s, partition = %d%n",
                        record.offset(), record.key(), record.value(), record.partition());
                }
            }
        } finally {
            consumer.close();
        }
    }
}

手动提交偏移量

精确控制提交策略

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;
public class ManualCommitConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "manual-commit-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 禁止自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 设置批量大小
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                // 处理所有消息
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                }
                // 手动同步提交
                consumer.commitSync();
                // 或者手动异步提交 with callback
                // consumer.commitAsync((offsets, exception) -> {
                //     if (exception != null) {
                //         System.err.println("Commit failed for offsets " + offsets);
                //     }
                // });
            }
        } finally {
            try {
                // 最后同步提交，确保提交成功
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }
    }
    private static void processRecord(ConsumerRecord<String, String> record) {
        // 业务处理逻辑
        System.out.println("Processing: " + record.value());
    }
}

精准一次性消费模式

结合事务的消费者

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.time.Duration;
import java.util.*;
public class ExactlyOnceConsumer {
    private static KafkaProducer<String, String> createProducer() {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "consumer-transaction-1");
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        producer.initTransactions();
        return producer;
    }
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "exactly-once-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("input-topic"));
        KafkaProducer<String, String> producer = createProducer();
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                // 开启事务
                producer.beginTransaction();
                try {
                    // 处理消息并发送到输出主题
                    for (ConsumerRecord<String, String> record : records) {
                        String processedValue = process(record.value());
                        producer.send(new ProducerRecord<>("output-topic", 
                            record.key(), processedValue));
                    }
                    // 提交消费偏移量到事务
                    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                    for (ConsumerRecord<String, String> record : records) {
                        offsets.put(new TopicPartition(record.topic(), record.partition()),
                            new OffsetAndMetadata(record.offset() + 1));
                    }
                    producer.sendOffsetsToTransaction(offsets, "exactly-once-group");
                    // 提交事务
                    producer.commitTransaction();
                } catch (Exception e) {
                    // 回滚事务
                    producer.abortTransaction();
                    throw e;
                }
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }
    private static String process(String value) {
        // 业务处理
        return "processed: " + value;
    }
}

多线程消费者案例

使用线程池并发处理

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
public class MultiThreadConsumer {
    private static final int THREAD_COUNT = 4;
    private static final ExecutorService executor = Executors.newFixedThreadPool(THREAD_COUNT);
    private static final AtomicLong totalProcessed = new AtomicLong(0);
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "multi-thread-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000); // 5分钟
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (!records.isEmpty()) {
                    // 提交给线程池处理
                    List<Future<?>> futures = new ArrayList<>();
                    for (ConsumerRecord<String, String> record : records) {
                        Future<?> future = executor.submit(() -> {
                            try {
                                processRecord(record);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        });
                        futures.add(future);
                    }
                    // 等待所有任务完成
                    for (Future<?> future : futures) {
                        try {
                            future.get();
                        } catch (InterruptedException | ExecutionException e) {
                            e.printStackTrace();
                        }
                    }
                    // 同步提交
                    consumer.commitSync();
                    System.out.println("Total processed: " + totalProcessed.incrementAndGet());
                }
            }
        } finally {
            consumer.close();
            executor.shutdown();
        }
    }
    private static void processRecord(ConsumerRecord<String, String> record) {
        // 模拟耗时处理
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("Thread: " + Thread.currentThread().getName() + 
            " processing: " + record.value());
    }
}

指定分区消费

手动分配分区

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
public class PartitionAssignConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "partition-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 手动分配特定分区的特定偏移量
        TopicPartition partition0 = new TopicPartition("test-topic", 0);
        TopicPartition partition1 = new TopicPartition("test-topic", 1);
        consumer.assign(Arrays.asList(partition0, partition1));
        // 从最新偏移量开始消费
        // consumer.seekToBeginning(Arrays.asList(partition0, partition1));
        // 或从指定偏移量开始消费
        consumer.seek(partition0, 100);  // 从分区0的第100条消息开始
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("Partition: %d, Offset: %d, Key: %s, Value: %s%n",
                        record.partition(), record.offset(), record.key(), record.value());
                }
                // 异步提交
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        System.err.println("Commit failed: " + exception.getMessage());
                    }
                });
            }
        } finally {
            consumer.close();
        }
    }
}

消费者组动态分配监控

带消费者组信息的客户端

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
public class ConsumerGroupInfoExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "monitoring-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 设置消费者组重平衡监听器
        consumer.subscribe(Arrays.asList("test-topic"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                System.out.println("Partitions revoked: " + partitions);
                // 在重平衡前保存偏移量
                consumer.commitSync();
            }
            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                System.out.println("Partitions assigned: " + partitions);
                // 可以在这里从外部存储恢复偏移量
            }
        });
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                // 获取当前消费者组的分配信息
                if (!records.isEmpty()) {
                    System.out.println("Group info: " + consumer.groupMetadata());
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.printf("Received: topic=%s partition=%d offset=%d key=%s value=%s%n",
                            record.topic(), record.partition(), record.offset(), 
                            record.key(), record.value());
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }
}

实战：带重试和死信队列的消费者

企业级消息处理模式

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import java.time.Duration;
import java.util.*;
public class EnterpriseConsumer {
    private static final int MAX_RETRIES = 3;
    private static final String DLQ_TOPIC = "test-topic-dlq";
    public static void main(String[] args) {
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "enterprise-group");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
        KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
        consumer.subscribe(Arrays.asList("test-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        processWithRetry(record);
                    } catch (Exception e) {
                        System.err.println("Processing failed after retries: " + e.getMessage());
                        // 发送到死信队列
                        producer.send(new ProducerRecord<>(DLQ_TOPIC, record.key(), record.value()));
                        producer.flush();
                    }
                }
                // 确认所有消息已处理
                consumer.commitSync();
            }
        } finally {
            consumer.close();
            producer.close();
        }
    }
    private static void processWithRetry(ConsumerRecord<String, String> record) throws Exception {
        Exception lastException = null;
        for (int retry = 0; retry < MAX_RETRIES; retry++) {
            try {
                // 模拟业务处理
                System.out.println("Attempt " + (retry + 1) + ": processing " + record.value());
                processMessage(record.value());
                return; // 成功则返回
            } catch (Exception e) {
                lastException = e;
                System.err.println("Attempt " + (retry + 1) + " failed: " + e.getMessage());
                // 退避等待
                try {
                    Thread.sleep(1000 * (retry + 1));
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    throw new RuntimeException("Interrupted during retry", ie);
                }
            }
        }
        throw lastException;
    }
    private static void processMessage(String message) throws Exception {
        // 模拟可能导致失败的业务逻辑
        if (message.contains("error")) {
            throw new RuntimeException("Business logic error");
        }
        System.out.println("Successfully processed: " + message);
    }
}

性能优化案例

批量处理和高吞吐配置

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;
public class OptimizedConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "optimized-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 性能优化配置
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1000");
        props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "1024");
        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "500");
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, "1048576");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        // 禁用消费者组协调
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
            "org.apache.kafka.clients.consumer.RoundRobinAssignor");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("high-volume-topic"));
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    continue;
                }
                // 批量处理
                Map<Integer, List<ConsumerRecord<String, String>>> byPartition = 
                    records.records(new TopicPartition("high-volume-topic", 0));
                // 按分区批量处理
                records.partitions().forEach(partition -> {
                    List<ConsumerRecord<String, String>> partitionRecords = 
                        records.records(partition);
                    // 批量处理逻辑
                    List<String> batchValues = partitionRecords.stream()
                        .map(ConsumerRecord::value)
                        .collect(Collectors.toList());
                    // 批量保存到数据库等
                    batchSave(batchValues);
                });
                // 自动提交由Kafka处理
            }
        } finally {
            consumer.close();
        }
    }
    private static void batchSave(List<String> values) {
        // 批量保存到数据库
        System.out.println("Batch saving " + values.size() + " records");
    }
}

关键配置说明

配置项 默认值 说明
bootstrap.servers Kafka服务器地址
group.id 消费者组ID
enable.auto.commit true 是否自动提交偏移量
auto.commit.interval.ms 5000 自动提交间隔
auto.offset.reset latest 偏移量重置策略
max.poll.records 500 每次poll最大记录数
max.poll.interval.ms 300000 最大poll间隔
session.timeout.ms 10000 会话超时时间
isolation.level read_uncommitted 事务隔离级别
  1. 使用异步提交时：确保最终能提交成功
  2. 重平衡处理：在重平衡前保存偏移量
  3. 异常处理：为不可恢复的异常准备DLQ
  4. 性能优化：合理设置批量大小和拉取间隔
  5. 监控：记录消费者组的lag等信息

这些案例覆盖了Kafka Consumer的主要使用场景，你可以根据实际需求选择合适的模式，需要我详细解释某个特定场景吗？

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 足球直播 足球直播 足球直播 篮球直播 篮球直播 篮球直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 亚冠直播 亚冠直播 亚冠直播 欧冠直播 欧冠直播 欧冠直播 英超直播 英超直播 英超直播 西甲直播 西甲直播 西甲直播 中超直播 中超直播 中超直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 曼联直播 曼联直播 曼联直播 日职联直播 日职联直播 日职联直播 澳超直播 澳超直播 澳超直播 阿森纳直播 阿森纳直播 阿森纳直播 足球直播 足球直播 足球直播 NBA直播 NBA直播 NBA直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 欧冠直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 篮球直播 篮球直播 篮球直播 篮球直播 篮球直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 中超直播 中超直播 中超直播 中超直播 中超直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 篮球直播 篮球直播 篮球直播 篮球直播 篮球直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 中超直播 中超直播 中超直播 中超直播 中超直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播 中超直播

上一篇Supplier案例

下一篇Java函数式接口案例

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝