大数据学习(四) kafka安装配置

本文主要是介绍大数据学习(四) kafka安装配置，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

大数据学习（四）kafka安装配置

- - - 概述
    - - 定义
      - 消息队列好处
      - 消息队列的两种模式
    - kafka安装配置
    - - 解压
      - 配置
      - 启动集群
    - kafka命令行操作
    - - 查看当前服务器中所有topic
      - 创建topic
      - 删除 topic
      - 发送消息
      - 消费消息
    - kafka API
    - - 异步发送API
      - 导入依赖
        编写代码
      - 同步发送API
      - 自动提交offset
      - 导入依赖
        编写代码
      - 手动提交offset
      - 同步提交offset
        异步提交offset
      - 自定义存储offset
    - 自定义Interceptor
    - - 增加时间戳拦截器
      - 统计发送消息成功和发送失败消息数，并在 producer 关闭时打印这两个计数器
      - producer 主程序

概述

定义

Kafka 是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域。

消息队列好处

解耦允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

可恢复性 系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

缓冲有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

灵活性 & 峰值处理能力 在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

异步通信 很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

消息队列的两种模式

点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除）
消息生产者生产消息发送到Queue中，然后消息消费者从Queue中取出并且消费消息。消息被消费以后，queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息。 Queue 支持存在多个消费者，但是对一个消息而言，只会有一个消费者可以消费。
发布/订阅模式（一对多，消费者消费数据之后不会清除消息）
消息生产者（发布）将消息发布到 topic 中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到 topic 的消息会被所有订阅者消费

kafka安装配置

解压

tar -zxvf  kafka_2.11-0.11.0.0.tgz -C /opt/module
cd ../module/
mv kafka_2.11-0.11.0.0/ kafka

在kafka安装目录下创建logs文件夹（此logs非彼logs，此logs是数据存放目录，非日志文件目录）
mkdir logs

配置

修改配置文件

cd config
vi server.properties

修改为：

#broker 的全局唯一编号，不能重复
broker.id=0
#删除 topic 功能使能
delete.topic.enable=true
#处理网络请求的线程数量
num.network.threads=3
#用来处理磁盘 IO 的现成数量
num.io.threads=8
#发送套接字的缓冲区大小
socket.send.buffer.bytes=102400
#接收套接字的缓冲区大小
socket.receive.buffer.bytes=102400
#请求套接字的缓冲区大小
socket.request.max.bytes=104857600
#kafka 运行日志存放的路径
log.dirs=/opt/module/kafka/logs
#topic 在当前 broker 上的分区个数
num.partitions=1
#用来恢复和清理 data 下数据的线程数量
num.recovery.threads.per.data.dir=1
#segment 文件保留的最长时间，超时将被删除
log.retention.hours=168
#配置连接 Zookeeper 集群地址
zookeeper.connect=master:2181,slave1:2181,沙slave2:2181

配置环境变量

# 配置环境变量

vi /etc/profile
# KAFKA_HOME
export KAFKA_HOME=/opt/module/kafka
export PATH=$PATH:$KAFKA_HOME/bin

最后别忘source /etc/profile
将安装目录分发到两外两台机器
分别在 slave1和 slave2 上修改配置文件config/server.properties
中的 broker.id=1、broker.id=2
注：broker.id 不得重复

启动集群

依次在三台机器启动kafka

kafka-server-start.sh -daemon config/server.properties

kafka命令行操作

查看当前服务器中所有topic

kafka-topics.sh --zookeeper master:9092 --list

创建topic

kafka-topics.sh --zookeeper master:9092 --create --replication-factor 3 --partitions 1 --topic firstTopic

# --topic 定义topic名
# --replication-factor 定义副本数
# --partitions 分区数

删除 topic

kafka-topics.sh --zookeeper 
master:9092 --delete --topic firstTopic

需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除。

发送消息

kafka-console-producer.sh --broker-list master:9092 --topic firstTopic

消费消息

kafka-console-consumer.sh \
--zookeeper master:9092 --topic firstTopic

kafka API

异步发送API

导入依赖

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

编写代码

1、不带回调函数

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducer {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties props = new Properties();
        //kafka 集群，broker-list
        props.put("bootstrap.servers", "master:9092");
        props.put("acks", "all");
        //重试次数
        props.put("retries", 1);
        //批次大小
        props.put("batch.size", 16384);
        //等待时间
        props.put("linger.ms", 1);
        //RecordAccumulator 缓冲区大小
        props.put("buffer.memory", 33554432);
        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++) {
            producer.send(new ProducerRecord < String, String > ("first", Integer.toString(i), Integer.toString(i)));
        }
        producer.close();
    }
}

2、带回调函数的API
回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是
RecordMetadata 和 Exception，如果 Exception 为 null，说明消息发送成功，如果
Exception 不为 null，说明消息发送失败。

import org.apache.kafka.clients.producer.*;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducer {
    public static void main(String[] args) throws ExecutionException,
        InterruptedException {
            Properties props = new Properties();
            props.put("bootstrap.servers", "master:9092"); //kafka 集群， broker - list
            props.put("acks", "all");
            props.put("retries", 1); //重试次数
            props.put("batch.size", 16384); //批次大小
            props.put("linger.ms", 1); //等待时间
            props.put("buffer.memory", 33554432); //RecordAccumulator 缓冲区大小
            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++) {
                producer.send(new ProducerRecord < String, String > ("first", Integer.toString(i), Integer.toString(i)), new Callback() {
                    //回调函数，该方法会在 Producer 收到 ack 时调用，为异步调用
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        if(exception == null) {
                            System.out.println("success->" + metadata.offset());
                        }
                        else {
                            exception.printStackTrace();
                        }
                    }
                });
            }
            producer.close();
        }
}

同步发送API

同步发送的意思就是，一条消息发送之后，会阻塞当前线程，直至返回 ack。
由于 send 方法返回的是一个 Future 对象，根据 Futrue 对象的特点，我们也可以实现同
步发送的效果，只需在调用 Future 对象的 get 方发即可。

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducer {
    public static void main(String[] args) throws ExecutionException,
        InterruptedException {
            Properties props = new Properties();
            props.put("bootstrap.servers", "master:9092"); //kafka 集群， broker - list
            props.put("acks", "all");
            props.put("retries", 1); //重试次数
            props.put("batch.size", 16384); //批次大小
            props.put("linger.ms", 1); //等待时间
            props.put("buffer.memory", 33554432); //RecordAccumulator 缓冲区大小
            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++) {
                producer.send(new ProducerRecord < String, String > ("first", Integer.toString(i), Integer.toString(i))).get();
            }
            producer.close();
        }
}

自动提交offset

导入依赖

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

编写代码

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class CustomConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "master: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);
        consumer.subscribe(Arrays.asList("first"));
        while(true) {
            ConsumerRecords < String, String > records = consumer.poll(100);
            for(ConsumerRecord < String, String > record: records) System.out.printf("offset = %d, key = %s, value  = % s % n ", record.offset(), record.key(), record.value());
            }
        }
    }

手动提交offset

虽然自动提交 offset 十分简介便利，但由于其是基于时间提交的，开发人员难以把握
offset 提交的时机。因此 Kafka 还提供了手动提交 offset 的 API。
手动提交 offset 的方法有两种：分别是 commitSync（同步提交）和 commitAsync（异步
提交）。两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交；不同点是，
commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，
也会出现提交失败）；而 commitAsync 则没有失败重试机制，故有可能提交失败。

同步提交offset

由于同步提交 offset 有失败重试机制，故更加可靠，以下为同步提交 offset 的示例。

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Arrays;
import java.util.Properties;
public class CustomComsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //Kafka 集群
        props.put("bootstrap.servers", "master:9092");
        //消费者组，只要 group.id 相同，就属于同一个消费者组
        props.put("group.id", "test");
        props.put("enable.auto.commit", "false"); //关闭自动提交 offset
        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);
        consumer.subscribe(Arrays.asList("first")); //消费者订阅主题
        while(true) {
            //消费者拉取数据
            ConsumerRecords < String, String > records = consumer.poll(100);
            for(ConsumerRecord < String, String > record: records) {
                System.out.printf("offset = %d, key = %s, value  = % s % n ", record.offset(), record.key(), record.value());
                    }
                    //同步提交，当前线程会阻塞直到 offset 提交成功
                consumer.commitSync();
            }
        }
    }

异步提交offset

虽然同步提交 offset 更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞
吐量会收到很大的影响。因此更多的情况下，会选用异步提交 offset 的方式。

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
public class CustomConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        //Kafka 集群
        props.put("bootstrap.servers", "master:9092");
        //消费者组，只要 group.id 相同，就属于同一个消费者组
        props.put("group.id", "test");
        //关闭自动提交 offset
        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);
        consumer.subscribe(Arrays.asList("first")); //消费者订阅主题
        while(true) {
            ConsumerRecords < String, String > records = consumer.poll(100); //消费者拉取数据
            for(ConsumerRecord < String, String > record: records) {
                System.out.printf("offset = %d, key = %s, value  = % s % n ", record.offset(), record.key(), record.value());
                    }
                    //异步提交
                consumer.commitAsync(new OffsetCommitCallback() {@
                    Override
                    public void onComplete(Map < TopicPartition, OffsetAndMetadata > offsets, Exception exception) {
                        if(exception != null) {
                            System.err.println("Commit failed for" + offsets);
                        }
                    }
                });
            }
        }
    }

自定义存储offset

Kafka 0.9 版本之前，offset 存储在 zookeeper，0.9 版本及之后，默认将 offset 存储在 Kafka的一个内置的 topic 中。除此之外，Kafka 还可以选择自定义存储 offset。

offset 的维护是相当繁琐的，因为需要考虑到消费者的 Rebalace。

当有新的消费者加入消费者组、已有的消费者推出消费者组或者所订阅的主题的分区发生变化，就会触发到分区的重新分配，重新分配的过程叫做 Rebalance。

消费者发生 Rebalance 之后，每个消费者消费的分区就会发生变化。因此消费者要首先获取到自己被重新分配到的分区，并且定位到每个分区最近提交的 offset 位置继续消费。

要实现自定义存储 offset，需要借助 ConsumerRebalanceListener，以下为示例代码，其中提交和获取 offset 的方法，需要根据所选的 offset 存储系统自行实现。

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;
public class CustomConsumer {
    private static Map < TopicPartition, Long > currentOffset = new HashMap < > ();
    public static void main(String[] args) {
        //创建配置信息
        Properties props = new Properties();
        //Kafka 集群
        props.put("bootstrap.servers", "master:9092");
        //消费者组，只要 group.id 相同，就属于同一个消费者组
        props.put("group.id", "test");
        //关闭自动提交 offset
        props.put("enable.auto.commit", "false");
        //Key 和 Value 的反序列化类
        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);
        //消费者订阅主题
        consumer.subscribe(Arrays.asList("first"), new ConsumerRebalanceListener() {
            //该方法会在 Rebalance 之前调用
            @Override
            public void
            onPartitionsRevoked(Collection < TopicPartition > partitions) {
                    commitOffset(currentOffset);
                }
                //该方法会在 Rebalance 之后调用
                @Override
            public void
            onPartitionsAssigned(Collection < TopicPartition > partitions) {
                currentOffset.clear();
                for(TopicPartition partition: partitions) {
                    consumer.seek(partition, getOffset(partition)); //定位到最近提交的 offset 位置继续消费
                }
            }
        });
        while(true) {
            ConsumerRecords < String, String > records = consumer.poll(100); //消费者拉取数据
            for(ConsumerRecord < String, String > record: records) {
                System.out.printf("offset = %d, key = %s, value  = % s % n ", record.offset(), record.key(), record.value());
                    currentOffset.put(new TopicPartition(record.topic(), record.partition()), record.offset());
                }
                commitOffset(currentOffset); //异步提交
            }
        }
        //获取某分区的最新 offset
        private static long getOffset(TopicPartition partition) {
                return 0;
            }
            //提交该消费者所有分区的 offset
        private static void commitOffset(Map < TopicPartition, Long > currentOffset) {}
    }

自定义Interceptor

实现一个简单的双 interceptor 组成的拦截链。第一个 interceptor 会在消息发送前将时间
戳信息加到消息 value 的最前部；第二个 interceptor 会在消息发送后更新成功发送消息数或
失败发送消息数。

增加时间戳拦截器

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class TimeInterceptor implements
        ProducerInterceptor<String, String> {
    @Override
    public void configure(Map<String, ?> configs) {
    }

    @Override
    public ProducerRecord<String, String>
    onSend(ProducerRecord<String, String> record) {
// 创建一个新的 record，把时间戳写入消息体的最前部
        return new ProducerRecord(record.topic(),
                record.partition(), record.timestamp(), record.key(),
                System.currentTimeMillis() + "," +
                        record.value().toString());
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata,
                                  Exception exception) {
    }

    @Override
    public void close() {
    }
}

统计发送消息成功和发送失败消息数，并在 producer 关闭时打印这两个计数器

import java.util.Map;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class CounterInterceptor implements
        ProducerInterceptor<String, String> {
    private int errorCounter = 0;
    private int successCounter = 0;

    @Override
    public void configure(Map<String, ?> configs) {
    }

    @Override
    public ProducerRecord<String, String>
    onSend(ProducerRecord<String, String> record) {
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata,
                                  Exception exception) {
// 统计成功和失败的次数
        if (exception == null) {
            successCounter++;
        } else {
            errorCounter++;
        }
    }

    @Override
    public void close() {
        // 保存结果
        System.out.println("Successful sent: " + successCounter);
        System.out.println("Failed sent: " + errorCounter);
    }
}

producer 主程序

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

public class InterceptorProducer {
    public static void main(String[] args) throws Exception {
// 1 设置配置信息
        Properties props = new Properties();
        props.put("bootstrap.servers", "master:9092");
        props.put("acks", "all");
        props.put("retries", 3);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
// 2 构建拦截链
        List<String> interceptors = new ArrayList<>();
        interceptors.add("com.atguigu.kafka.interceptor.TimeInterceptor");
        interceptors.add("com.atguigu.kafka.interceptor.CounterInterceptor");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
                interceptors);
        String topic = "first";
        Producer<String, String> producer = new
                KafkaProducer<>(props);
// 3 发送消息
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new
                    ProducerRecord<>(topic, "message" + i);
            producer.send(record);
        }
// 4 一定要关闭 producer，这样才会调用 interceptor 的 close 方法
        producer.close();
    }
}

这篇关于大数据学习(四) kafka安装配置的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！