Kafka指南

缺乏、安全感 2022-12-27 08:55 178阅读 0赞

> ## Kafka入门、API、Spring集成 ##

# 入门 #

## 下载代码 ##

[https://kafka.apache.org/downloads][https_kafka.apache.org_downloads]

kafka\_2.13-2.6.0.tgz

## 启动服务器 ##

创建一个单节点ZooKeeper实例

`bin``/zookeeper-server-start``.sh config``/zookeeper``.properties`

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70][]

`启动Kafka服务器`

`bin/kafka-server-start.sh config/server.properties`

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 1][]

## `创建一个 topic` ##

`创建一个名为“test”的topic，它有一个分区和一个副本`

`bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test`  
![20201210150919378.png][]

运行list（列表）命令来查看这个topic

![20201210151002460.png][]

## 发送一些消息 ##

运行 producer，然后在控制台输入一些消息以发送到服务器

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

This is a message

This is another message

![20201210151156573.png][]

## 启动一个 consumer ##

将消息转储到标准输出

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

![20201210151306367.png][]

## 设置多代理集群 ##

cp config/server.properties config/server-1.properties

cp config/server.properties config/server-2.properties

![20201210151504297.png][]

config/server-1.properties:
        broker.id=1
        listeners=PLAINTEXT://:9093
        log.dir=/tmp/kafka-logs-1
     
    config/server-2.properties:
        broker.id=2
        listeners=PLAINTEXT://:9094
        log.dir=/tmp/kafka-logs-2

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 2][]

已经建立Zookeeper和一个单节点了，现在我们只需要启动两个新的节点

bin/kafka-server-start.sh config/server-1.properties &

bin/kafka-server-start.sh config/server-2.properties &

bin/zookeeper-server-start.sh config/zookeeper.properties &
    bin/kafka-server-start.sh config/server.properties &
    bin/kafka-server-start.sh config/server-1.properties &
    bin/kafka-server-start.sh config/server-2.properties &

![20201210153452372.png][]

创建一个副本为3的新topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

运行"describe topics"命令来查看代理

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic

![20201210153656163.png][]

示例中，节点1是该主题中唯一分区的领导者

发表一些信息给我们的新topic

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

my test message 1

my test message 2

消费这些消息

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 3][]

测试一下容错性。Broker 1 现在是 leader，让我们来杀了它

ps aux | grep server-1.properties

![2020121016185353.png][]

kill -9 6369

领导权已经切换到一个从属节点，而且节点1也不在同步副本集中了

bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic

![20201210162231588.png][]

# API #

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

## 生产者 ##

public class KafkaProducerTest {
        public static void main(String[] args) throws Exception {
            /**
             * 1、准备配置信息
             */
            Properties props = new Properties();
            //定义Kafka服务器地址列表，不需要指定所有的broker
            props.put("bootstrap.servers", "localhost:9092");
            //生产者需要leader确认请求完成之前接收的应答数
            props.put("acks", "-1");
            //客户端失败重试次数
            props.put("retries", 1);
            //生产者打包消息的批量大小，以字节为单位，此次是16k
            props.put("batch.size", 16384);
            //生产者延迟1ms发送消息
            props.put("linger.ms", 1);
            //生产者缓存内存的大小，以字节为单位，此处是32m
            props.put("buffer.memory", 33554432);
            // key 序列化类
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            // value 序列化类
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            /**
             * 2、创建生产者对象
             */
            KafkaProducer<String, String> producer = new KafkaProducer<>(props);
            /**
             * 3、通过生产者发送消息
             *
             */
            /**
             * 发送消息的三种方式：
             *
             * - 同步阻塞发送
             * 适用场景：发送消息不能出错，消息的顺序不能乱，不关心高吞吐量。
             *
             * - 异步发送（发送并忘记）
             * 适用场景：发送消息不管会不会出错，消息的顺序乱了没有关系，关心高吞吐量。
             *
             * - 异步发送（进行回调处理）
             * 适用场景：发送消息不能出错，但我不关心消息的具体顺序。
             */
            // 1、同步阻塞发送
    //        sync(producer);
            // 2、异步发送（发送并忘记）
            //async1(producer);
            // 3、异步发送（回调函数）
            async2(producer);
        }
    
        /**
         * 同步阻塞发送
         *
         * @param producer
         * @throws Exception
         */
        private static void sync(KafkaProducer producer) throws Exception {
            System.out.println("同步发送消息start......");
            ProducerRecord<String, String> record = new ProducerRecord<>("alanchen_topic", 0, "key-sync", "同步发送消息");
            Future<RecordMetadata> send = producer.send(record);
            RecordMetadata recordMetadata = send.get();
            producer.flush();
            System.out.println(recordMetadata);
            System.out.println("同步发送消息end......");
        }
    
        /**
         * 异步发送（发送并忘记）
         *
         * @param producer
         * @throws Exception
         */
        private static void async1(KafkaProducer producer) {
            System.out.println("异步发送（发送并忘记）start......");
            ProducerRecord<String, String> record = new ProducerRecord<>("alanchen_topic", 0, "key-async1", "异步发送（发送并忘记）");
            producer.send(record);
            producer.flush();
            System.out.println("异步发送（发送并忘记）end......");
        }
    
        /**
         * 异步发送（回调函数）
         *
         * @param producer
         * @throws Exception
         */
        private static void async2(KafkaProducer producer) {
            System.out.println("异步发送（回调函数）start......");
            ProducerRecord<String, String> record = new ProducerRecord<>("alanchen_topic", 0, "key-async2", "异步发送（回调函数）");
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    System.out.println("异步发送消息成功：" + metadata);
                    if (exception != null) {
                        exception.printStackTrace();
                    }
                }
            });
            producer.flush();
            System.out.println("异步发送（回调函数）end......");
        }
    }

## 消费者 ##

public class KafkaConsumerTest {
        public static void main(String[] args) {
            /**
             * 1、准备配置信息
             */
            Properties props = new Properties();
            //定义Kafka服务器地址列表，不需要指定所有的broker
            props.put("bootstrap.servers", "localhost:9092");
            //消费者组ID
            props.put("group.id", "ac");
            //是否自动确认offset
            props.put("enable.auto.commit", "true");
            //自动确认offset时间间隔
            props.put("auto.commit.interval.ms", 1000);
            //生产者延迟1ms发送消息
            props.put("linger.ms", 1);
            //生产者缓存内存的大小，以字节为单位，此处是32m
            props.put("buffer.memory", 33554432);
            // key 序列化类
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            // value 序列化类
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            /**
             * 2、创建一个消费者
             */
            KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
            /**
             * 3、指定消费哪一个topic的数据
             */
            //指定分区消费
            TopicPartition partition = new TopicPartition("alanchen_topic", 0);
            //获取已经提交的偏移量
            long offset = 0L;
            OffsetAndMetadata offsetAndMetadata = consumer.committed(partition);
            if (offsetAndMetadata != null) {
                offset = offsetAndMetadata.offset();
            }
            System.out.println("当前消费的偏移量：" + offset);
            // 指定偏移量消费
            consumer.assign(Arrays.asList(partition));
            consumer.seek(partition, offset);
            //循环拉取数据
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("消费的数据为：" + record.value());
                }
            }
        }
    }

# Spring集成 #

<dependency>
                <groupId>org.springframework.kafka</groupId>
                <artifactId>spring-kafka</artifactId>
            </dependency>

## 生产者 ##

@Component
    @Slf4j
    public class KafkaProvider {
        /**
         * TOPIC
         */
        private static final String TOPIC = "xiaoha";
        @Autowired
        private KafkaTemplate<String, String> kafkaTemplate;
    
        public void sendMessage(long orderId, String orderNum, LocalDateTime createTime) {
            Order order = Order.builder()
                    .orderId(orderId)
                    .orderNum(orderNum)
                    .createTime(createTime)
                    .build();
            // 发送消息
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(TOPIC, JSONObject.toJSONString(order));
            // 监听回调
            future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                @Override
                public void onFailure(Throwable throwable) {
                    log.info("## Send message fail ...");
                }
    
                @Override
                public void onSuccess(SendResult<String, String> result) {
                    log.info("## Send message success ...");
                }
            });
        }
    }

## 消费者 ##

@Component
    @Slf4j
    public class KafkaConsumer {
        @KafkaListener(topics = "xiaoha", groupId = "group_id")
        public void consume(String message) {
            log.info("## consume message: {}", message);
        }
    }

## 测试 ##

@RunWith(SpringRunner.class)
    @ComponentScan(basePackages = {"com.moses.microcloud.platform"})
    @SpringBootTest(classes = SpringBootKafkaApplicationTests.class)
    public class SpringBootKafkaApplicationTests {
        @Autowired
        private KafkaProvider kafkaProvider;
    
        @Test
        public void sendMessage() throws InterruptedException {
            // 发送 1000 个消息
            for (int i = 0; i < 10; i++) {
                long orderId = i + 1;
                String orderNum = UUID.randomUUID().toString();
                kafkaProvider.sendMessage(orderId, orderNum, LocalDateTime.now());
            }
            TimeUnit.MINUTES.sleep(1);
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 4][]

[https_kafka.apache.org_downloads]: https://kafka.apache.org/downloads
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70]: /images/20221120/afb127b658654d0d8dae44e170f6cb22.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 1]: /images/20221120/de85a3d6055b48138d10e58be0f059cf.png
[20201210150919378.png]: /images/20221120/6da8534feec0461295e06dd79c7c95d3.png
[20201210151002460.png]: /images/20221120/055b87aa57524e3fb8122395bba87599.png
[20201210151156573.png]: /images/20221120/29388659e23c48af92cc9fde560e4485.png
[20201210151306367.png]: /images/20221120/86b8a35ccc784be98976b082e8282f27.png
[20201210151504297.png]: /images/20221120/d8e34d3972b345fc9a110a662b3d8828.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 2]: /images/20221120/b50ad7b6a7914b15b690dc4a620706eb.png
[20201210153452372.png]: /images/20221120/78cb1c57e272487484e4ba30d4e14e43.png
[20201210153656163.png]: /images/20221120/b5969145a36343dea1c4ea2482531abf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 3]: /images/20221120/2a5dd572368f4e56a5126b5f39e585d1.png
[2020121016185353.png]: /images/20221120/49d3152ba6b54e139b61266ef6d4e026.png
[20201210162231588.png]: /images/20221120/f07306755e1741fdb1fe364193e8920f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3N0ZXZlX2ZyYW5r_size_16_color_FFFFFF_t_70 4]: /images/20221120/37b6c8c017474ffcbc1856766019871c.png