kafka - Kafka - clients - 《Big Data》

Kafka_Java
Spring-Kafka

Kafka_Java

简介

[kafka-clients]
    ：http://kafka.apache.org/0100/javadoc/index.html # 官方文档
// 手动创建 topic

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.2.0</version>
</dependency>

[核心AP]

    ：Producer    # 生产者，发布消息 ，线程安全
        => 由一个持有未发送消息记录的资源池和一个用来向Kafka集群发送消息记录的后台IO线程组成
    => 使用后需要关闭，否则导致资源泄漏
    => send方法是异步的，添加到待发送缓冲区并立即返回

    ：Consumer    # 消费者，消费消息，非线程安全
      => High-level API # 提供了高度抽象的API
    => Low-level API  # 提供了更强的控制能力，但使用起来较为繁琐

    ：Streams

    ：Connect

    ：Properties # 属性参数

    ：KafkaConsumer 消费者
      => ConsumerRecords

  ：KafkaProducer 生产者
      => Producer
      => ProducerRecord

Producer 生产者

[Producer]

    // 创建Topic
    kafka-topics.sh
        --create
        --zookeeper localhost:2181
        --replication-factor 1
        --partitions 1 --topic kafkaTopic

    // 服务器端开启消费者，进行测试
    kafka-console-consumer.sh 
        --bootstrap-server 47.110.253.133:9092
        --topic kafkaTopic
        --from-beginning

[Producer]

// topic , 手动在服务器创建, 并开启消费着消费
String topic = "kafkaTopic";

// 设置属性
Properties props = new Properties();
// kafka 集群地址
props.put("bootstrap.servers", "47.110.253.133:9092");

props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);

// 指定key value 序列化方式
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

// 初始化生产者
Producer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++) {
  // 推送消息
  Thread.sleep(1000);
  producer.send(new ProducerRecord(topic, Integer.toString(i), Integer.toString(i)));
}

// 关闭资源
producer.close();

[参数属性]

        => bootstrap.servers 配置项处需要填写我们要发送到的Kafka集群地址
    => ack 配置项用来控制producer要求leader确认多少消息后返回调用成功,当值为0时producer不需要等待任何确认消息
       当值为1时只需要等待leader确认。当值为-1或all时需要全部ISR集合返回确认才可以返回成功
        => retries > 0 时，如果发送失败，会自动尝试重新发送数据

    => buffer.memory、batch.size、linger.ms 三个参数用来控制缓冲区大小和延迟发送时间
        => key.serializer 和 value.serializer 指定使用什么序列化方式将用户提供的key和value进行序列化

Consumer 消费者

[Consumer]

// 手动创建 topic
String topic = "kafkaTopic";

// 配置消费者参数属性
Properties props = new Properties();
// kafka 集群
props.put("bootstrap.servers", "47.110.253.133:9092");

props.put("group.id", "test");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");

// 序列化
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// 定义消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅 topic
consumer.subscribe(Arrays.asList(topic));
while (true) {
    // 消费消息
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        Thread.sleep(1000);
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}

[参数属性]

  => enable.auto.commit配置项指定了提交offset的方式为自动提交
  => auto.commit.interval.ms配置项配置了每次自动提交的时间间隔

[手动提交方式 方式]

// 手动创建 topic
String topic = "kafkaTopic";

// 配置消费者参数属性
Properties props = new Properties();
// kafka 集群
props.put("bootstrap.servers", "47.110.253.133:9092");

props.put("group.id", "test");
// 手动提交方式
props.put("enable.auto.commit", "false");

// 序列化
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

// 定义消费者
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅 topic
consumer.subscribe(Arrays.asList(topic));

final int minBatchSize = 2;
// 定义了一个ConsumerRecord的列表作为缓冲，当缓冲中的数据大于2条时，才一次性操作，并手动提交offset
List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
while (true) {
    // 消费消息
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        buffer.add(record);
    }

    if (buffer.size() >= minBatchSize) {
        // 业务操作, 入库
        for (ConsumerRecord<String, String> record : buffer) {
            Thread.sleep(1000);

            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        }

        // 手动提交，表示消息消费完成
        consumer.commitSync();

        buffer.clear();
    }
}

Spring-Kafka

spring - kafka 整合

[项目]

    ：https://spring.io/projects/spring-kafka

    ：https://yq.aliyun.com/articles/619556

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.2.0</version>
</dependency>

参数属性配置

######
## Kafka 服务配置
######

#brokers集群
bootstrap.servers=47.110.253.133:9092,192.168.11.40:9092
  # 即所有副本都同步到数据时send方法才返回, 以此来完全判断数据是否发送成功, 理论上来讲数据不会丢失
acks=all
  # 发送失败重试次数
retries=10
  # 批处理条数：当多个记录被发送到同一个分区时，生产者会尝试将记录合并到更少的请求中。这有助于客户端和服务器的性能
batch.size=1638
  # 批处理延迟时间上限：即1ms过后，不管是否达到批处理数，都直接发送一次请求
linger.ms=1
  # 32MB的批处理缓冲区
buffer.memory=33554432
  # 消费者群组ID，发布-订阅模式，即如果一个生产者，多个消费者都要消费，那么需要定义自己的群组，同一群组内的消费者只有一个能消费到消息
group.id=order-beta
  # 如果为true，消费者的偏移量将在后台定期提交。
enable.auto.commit=true
  # 如何设置为自动提交（enable.auto.commit=true），这里设置自动提交周期
auto.commit.interval.ms=1000
  # 在使用Kafka的组管理时，用于检测消费者故障的超时
session.timeout.ms=15000
  # 消费监听器容器并发数
concurrency = 3

KafkaBean配置

/**
 * Kafka 配置
 *
 * @author yuan
 */
@Configuration
public class KafkaConfig {

    /**
     * 配置 kafka 参数属性
     */
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();
        // kafka 集群 列表
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "47.110.253.133:9092");

        props.put(ProducerConfig.RETRIES_CONFIG, 0);
        props.put(ProducerConfig.ACKS_CONFIG, "-1");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 4096);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 40960);
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 5048576);

        // 序列化配置
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        return props;
    }

    /**
     * 获取工厂
     */
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    /**
     * 注册实例
     */
    @Bean(name = "kafkaTemplate")
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

生产者

@Resource
private KafkaTemplate kafkaTemplate;

/**
* 同步发送
*/
public void syncSend(String topic, String message) throws InterruptedException, ExecutionException, TimeoutException {
kafkaTemplate.send(topic, message).get(10, TimeUnit.SECONDS);
}

/**
* 异步发送
*/
public void asyncSend(String topic, String message) {

ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(topic, message);

// 重写发送，回调函数
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onSuccess(SendResult<Integer, String> result) {
  // 发送成功
  System.out.println("success");

  }

  @Override
  public void onFailure(Throwable ex) {
  // 发送失败
  System.out.println("failure");

  }
  });
  }

消费者

public class KafkaConsumerSerivceImpl implements MessageListener<String, String> {
    @Override
    public void onMessage(ConsumerRecord<String, String> data) {
        //根据不同的主题进行消费
        if("主题1".equals(data.topic())){

        }else if("主题2".equals(data.topic())){

        }
    }
}

@Component
public class KafkaConsumer {
    @KafkaListener(topics = {"testTopic"})
    public void receive(String message){
        System.out.println("消费消息:" + message);
    }
}

/**
 * Kafka 消费者配置
 *
 * @author yuan
 */
@Configuration
@EnableKafka
public class KafkaConsumerConfig {

    @Bean
    KafkaListenerContainerFactory
            <ConcurrentMessageListenerContainer<Integer, String>>
    kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory
                = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    /**
     * 消费者工厂
     */
    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    /**
     * 消费者配置
     */
    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        // Kafka 集群
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        // ...
        return props;
    }

}