ITViewing

Properties prop = new Properties();
prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
prop.put(ConsumerConfig.GROUP_ID_CONFIG, "testConsumer");
prop.put(ConsumerConfig.CLIENT_ID_CONFIG, "consumerDemo");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(prop);
consumer.subscribe(Collections.singleton("test"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, String> record : records) {
        String key = record.key();
        String value = record.value();
        System.err.println(record.toString());
    }
}

2021-08-22发表Middleware / Kafka4 分钟读完 (大约609个字)

Kafka 生产者客户端基本使用

必要的参数

bootstrap.servers

该参数为 broker 地址，不需要全部都填，因为 kafka 会从当前 broker 中获取其他 broker 信息。不过为了某个 broker 挂掉，一般填多个 broker 地址

key.serializer

2021-08-17发表Middleware / Kafka20 分钟读完 (大约2935个字)

Kafka

生产者发送消息的整体流程

消息追加器 `RecordAccumulator`

前面几个组件，在 3.1 的文章中，已经说清楚。现在来看 RecordAccumulator 组件

2021-08-10发表Middleware / Kafka6 分钟读完 (大约959个字)

Kafka 基本概念

kafka 应用场景

应用监控
网站用户行为追踪
流数据
持久性日志

基本概念

在说基本概念前，先看一下 kafka 的系统架构

2021-08-03发表Middleware / Kafka4 分钟读完 (大约589个字)

Kafka 搭建与实践

单机版 Docker 搭建

version: '2' 
services:
  zookeeper:
    image: wurstmeister/zookeeper
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka
    depends_on: [ zookeeper ]
    ports:
      - "9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ADVERTISED_PORT: 9092
      KAFKA_CREATE_TOPICS: "test:1:1"
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"