1：官网：
2：场景：
3：基本概念
4：安装与使用
启动kafka，运行日志在logs目录的server.log文件里
bin/kafka‐server‐start.sh ‐daemon config/server.properties #后台启动，不会打印日志到控制台
或者用
bin/kafka‐server‐start.sh config/server.properties &
停止kafka
12 bin/kafka‐server‐stop.sh
主题topic和消息日志log
集群模式
- 配置文件
broker.id属性在kafka集群中必须要是唯一
broker.id=1
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.65.60:9093
log.dir=/usr/local/data/kafka‐logs‐1
#kafka连接zookeeper的地址，要把多个kafka实例组成集群，对应连接的zookeeper必须相同
zookeeper.connect=192.168.65.60:2181
JAVA客户端访问
- 消费组、
Springboot整合
示例code

1：官网：

https://kafka.apache.org/documentation/
rocketMQ前身 kill me,，在kafka上开发，迭代成的

2：场景：

日志收集
收集各服务的日志
消息系统
解耦生产者，消费者，缓存消息
用户活动追踪
记录web用户和app用户活动，进行监控分析，或Hadoop 数据仓库离线分析
运营指标
收集分布式应用数据，报警和报告功能

3：基本概念

用到了Zk作为注册中心

名称	解释
Broker	消息中间件处理节点，一个Kafka节点就是一个broker ，一个或者多个Broker可以组成一个Kafka集群
Topic	Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic
Producer	消息生产者，向Broker发送消息的客户端
Consumer	消息消费者，从Broker读取消息的客户端
ConsumerGroup	每个Consumer属于一个特定的Consumer Group ，一条消息可以被多个不同的Consumer Group消费，但是一个 Consumer Group中只能有一个Consumer能够消费该消息
Partition	物理上的概念，一个topic可以分为多个partition ，每个 partition内部消息是有序的

服务端(brokers)和客户端(producer、consumer)之间通信通过TCP协议来完成

4：安装与使用

jvm环境

scala开发，需要jvm

依赖zookeeper

需要先安装好zk

下载

官网：https://kafka.apache.org/downloads

本地按照2.4.1 版本

wget https://mirror.bit.edu.cn/apache/kafka/2.4.1/kafka_2.11‐2.4.1.tgz # 2.11是scala的版本，2.4.1是kafka的版本 tar ‐xzf kafka_2.11‐2.4.1.tgz

修改配置

1 #broker.id属性在kafka集群中必须要是唯一 2 broker.id=0 3 #kafka部署的机器ip和提供服务的端口号 4 listeners=PLAINTEXT://192.168.65.60:9092 5 #kafka的消息存储文件 6 log.dir=/usr/local/data/kafka‐logs 7 #kafka连接zookeeper的地址 8 zookeeper.connect=192.168.65.60:2181

启动服务

启动kafka，运行日志在logs目录的server.log文件里
bin/kafka‐server‐start.sh ‐daemon config/server.properties #后台启动，不会打印日志到控制台
或者用
bin/kafka‐server‐start.sh config/server.properties &

bin/zkCli.sh
ls / #查看zk的根目录kafka相关节点
ls /brokers/ids #查看kafka节点

停止kafka
12 bin/kafka‐server‐stop.sh

server.properties核心配置

Property	Default	Description
broker.id	0	每个broker都可以用一个唯一的非负整数id进行标识；这个id可以作为broker的“ 你可以选择任意你喜欢的数字作为id，只要id是唯一的即可。
log.dirs	/tmp/kafka-logs	kafka存放数据的路径。这个路径并不是唯一的，可以是多个，路径之间只需要使用隔即可；每当创建新partition时，都会选择在包含最少partitions的路径下进行。
listeners	PLAINTEXT://192.168.65.60:909 2	server接受客户端连接的端口，ip配置kafka本机ip即可
zookeeper.connect	localhost:2181	zooKeeper连接字符串的格式为：hostname:port，此处hostname和port分别是 ZooKeeper集群中某个节点的host和port；zookeeper如果是集群，连接方式为 hostname1:port1, hostname2:port2, hostname3:port3

log.retention.hours	168	每个日志文件删除之前保存的时间。默认数据保存时间对所有topic都一样。
num.partitions	1	创建topic的默认分区数
default.replication.factor	1	自动创建topic的默认副本数量，建议设置为大于等于2
min.insync.replicas	1	当producer设置acks为-1时，min.insync.replicas指定replicas的最小数目（必须个repica的写数据都是成功的），如果这个数目没有达到，producer发送消息会产
delete.topic.enable	false	是否允许删除主题

创建主题

bin/kafka‐topics.sh ‐‐create ‐‐zookeeper 192.168.65.60:2181 ‐‐replication‐factor 1 ‐‐partitions 1 ‐‐topic test
创建主题名为 test , partitions分区只有一个

查看kafka主题

bin/kafka‐topics.sh ‐‐list ‐‐zookeeper 192.168.65.60:2181

删除主题

bin/kafka‐topics.sh ‐‐delete ‐‐topic test ‐‐zookeeper 192.168.65.60:2181

发送消息

启动生产者客户端，连接对应topic就可以发消息了 bin/kafka‐console‐producer.sh ‐‐broker‐list 192.168.65.60:9092 ‐‐topic test
> this is msg

this is anthor msg

消费消息

客户连接，启动即可，默认是消息最新消息 bin/kafka‐console‐consumer.sh ‐‐bootstrap‐server 192.168.65.60:9092 ‐‐topic test
如果要消费之前的消息 bin/kafka‐console‐consumer.sh ‐‐bootstrap‐server 192.168.65.60:9092 ‐‐from‐beginning ‐‐topic test

可以消费多个主题 bin/kafka‐console‐consumer.sh ‐‐bootstrap‐server 192.168.65.60:9092 ‐‐whitelist “test|test‐2”

单播消息

类型集群模式，一个消费组只有一个消费者消费同一条消息

bin/kafka‐console‐consumer.sh ‐‐bootstrap‐server 192.168.65.60:9092 ‐‐consumer‐property group.id=testGroup ‐‐topic test

多播消息

类似广播模式，同一个消息会被同一个消费组中的所有消费者消费
感觉没啥差别

查看消费组名：

bin/kafka‐consumer‐groups.sh ‐‐bootstrap‐server 192.168.65.60:9092 ‐‐list

查看消费偏移量：

bin/kafka‐consumer‐groups.sh ‐‐bootstrap‐server 192.168.65.60:9092 ‐‐describe ‐‐group testGroup

current-offset：当前消费组的已消费偏移量
log-end-offset：主题对应分区消息的结束偏移量(HW)
lag：当前消费组未消费的消息数

主题topic和消息日志log

topic：

同一个类别的消息发在同一个topic下，一个topic可以有多个分区

分区partition

一个有序的message序列，对应一个commit log文件
每个消息对应一个唯一的编号 offset ，
不同的partition的offset可能会重复

消息日志的保存时间一般一周，配置 log.retention.hours

每个consumer会基于自己的commit log中消费进度工作，offset由消费者维护

创建多个分区的主题

bin/kafka‐topics.sh ‐‐create ‐‐zookeeper 192.168.65.60:2181 ‐‐replication‐factor 1 ‐‐partitions 2 ‐‐topic test1

查看topic情况

bin/kafka‐topics.sh ‐‐describe ‐‐zookeeper 192.168.65.60:2181 ‐‐topic test1

leader节点负责给定partition的所有读写请求。
replicas 表示某个partition在哪几个broker上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr 是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点。

区别：

topic：同一个业务数据集，有的类似tag,，
topic可以划分多个partition，部署的不同的机器上
每个机器上的server就是broker

commit log 文件大小受机器文件系统大小限制，可以分配在不同的机器上，类似分布式存储消息，理论上可以分任意个分片
提高并行度

一个分组对一个消费组只能有一个消费者消费
一个消费者可以消费同一broker的所有分区

多个分区，多个消费组，

顺序消费只能是一个分区，才可以实现

集群模式

配置文件

cp config/server.properties config/server‐1.properties cp config/server.properties config/server‐2.properties

配置1：

broker.id属性在kafka集群中必须要是唯一
broker.id=1
#kafka部署的机器ip和提供服务的端口号
listeners=PLAINTEXT://192.168.65.60:9093
log.dir=/usr/local/data/kafka‐logs‐1
#kafka连接zookeeper的地址，要把多个kafka实例组成集群，对应连接的zookeeper必须相同
zookeeper.connect=192.168.65.60:2181

配置2：

broker.id=2
listeners=PLAINTEXT://192.168.65.60:9094
log.dir=/usr/local/data/kafka‐logs‐2
zookeeper.connect=192.168.65.60:2181

查看是否注册成功

登录zkclis

ls /brokers/ids

kafka很多信息记录在zk中，保证自己的无状态
broker没有主从之分
partitions有主从之分

集群消费：

针对每个partitions，都有一个broker起到leader，其他broker作为follower
broker没有主从之分
partitions有主从之分
leader处理这个partitions的读写亲戚，而follower被动复制leader的结果，不提供读写，保证一致性
如果对应的leader失效，其他follower会成为leader

Producers

生产者，负责吧Message发送到topic的哪个partition中，通过round-robin负载均衡

Consumer

传递模式：

queue队列模式：同一个消费组，多个consumer从服务器读取数据，消息只会被一个consumer消费
publish-subscribe模式，消息会被所有consumer消费

消费顺序

不太保证，
最佳方式只建一个partitions

API文档：
https://kafka.apache.org/documentation/#producerapi

JAVA客户端访问

maven依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka‐clients</artifactId> 
    <version>2.4.1</version> 
</dependency>

public class MsgProducer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Properties props = new Properties();
        //地址可以配一个broker就额可以了，但是高可用不行，还是多
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.26.128:9092");
         /*
         发出消息持久化机制参数
        （1）acks=0： 表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢消息。
        （2）acks=1： 至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一
             条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失。
        （3）acks=-1或all： 需要等待 min.insync.replicas(默认为1，推荐配置大于等于2) 这个参数配置的副本个数都成功写入日志，这种策略
            会保证只要有一个备份存活就不会丢失数据。这是最强的数据保证。一般除非是金融级别，或跟钱打交道的场景才会使用这种配置。
         */
        props.put(ProducerConfig.ACKS_CONFIG, "1");
         /*
        发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动，所以需要在
        接收者那边做好消息接收的幂等性处理
        */
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        //重试间隔设置
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
        //设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息发送性能，默认值是33554432，即32MB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        /*
        kafka本地线程会从缓冲区取数据，批量发送到broker，
        设置批量发送消息的大小，默认值是16384，即16kb，就是说一个batch满了16kb就发送出去
        */
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        /*
        默认值是0，意思就是消息必须立即被发送，但这样会影响性能
        一般设置10毫秒左右，就是说这个消息发送完后会进入本地的一个batch，如果10毫秒内，这个batch满了16kb就会随batch一起被发送出去
        如果10毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长
        */
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        //把发送的key从字符串序列化为字节数组
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把发送消息value从字符串序列化为字节数组
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        Producer<String, String> producer = new KafkaProducer<String, String>(props);
        int msgNum = 5;
        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);
        for (int i = 1; i <= msgNum; i++) {
            Order order = new Order(i, 100 + i, 1, 1000.00);
            //指定发送分区
            /*ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , 0, order.getOrderId().toString(), JSON.toJSONString(order));*/
            //未指定发送分区，具体发送的分区计算公式：hash(key)%partitionNum
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , order.getOrderId().toString(), JSON.toJSONString(order));
            //等待消息发送成功的同步阻塞方法
            RecordMetadata metadata = producer.send(producerRecord).get();
            System.out.println("同步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                    + metadata.partition() + "|offset-" + metadata.offset());
            //异步回调方式发送消息
            /*producer.send(producerRecord, new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("发送消息失败：" + exception.getStackTrace());
                    }
                    if (metadata != null) {
                        System.out.println("异步方式发送消息结果：" + "topic-" + metadata.topic() + "|partition-"
                                + metadata.partition() + "|offset-" + metadata.offset());
                    }
                    countDownLatch.countDown();
                }
            });*/
            //送积分 TODO
        }
        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.close();
    }
}

消费组、

public class MsgConsumer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";
    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.26.128:9092");
        // 消费分组名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        // 是否自动提交offset，默认就是true
        /*props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 自动提交offset的间隔时间
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");*/
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        /*
        当消费主题的是一个新的消费组，或者指定offset的消费方式，offset不存在，那么应该如何消费
        latest(默认) ：只消费自己启动之后发送到主题的消息
        earliest：第一次从头开始消费，以后按照消费offset记录继续消费，这个需要区别于consumer.seekToBeginning(每次都从头开始消费)
        */
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        /*
        consumer给broker发送心跳的间隔时间，broker接收到心跳如果此时有rebalance发生会通过心跳响应将
        rebalance方案下发给consumer，这个时间可以稍微短一点
        */
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        /*
        服务端broker多久感知不到一个consumer心跳就认为他故障了，会将其踢出消费组，
        对应的Partition也会被重新分配给其他consumer，默认是10秒
        */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
        //一次poll最大拉取消息的条数，如果消费者处理速度很快，可以设置大点，如果处理速度一般，可以设置小点
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 50);
        /*
        如果两次poll操作间隔超过了这个时间，broker就会认为这个consumer处理能力太弱，
        会将其踢出消费组，将分区分配给别的consumer消费
        */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 消费指定分区
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        //消息回溯消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*/
        //指定offset消费
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*/
        //从指定时间点开始消费
        /*List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        //从1小时前开始消费
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(TOPIC_NAME, par.partition()), fetchDataTime);
        }
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            //根据消费里的timestamp确定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }*/
        while (true) {
            /*
             * poll() API 是拉取消息的长轮询
             */
            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());
            }
            if (records.count() > 0) {
                // 手动同步提交offset，当前线程会阻塞直到offset提交成功
                // 一般使用同步提交，因为提交之后一般也没有什么逻辑代码了
                //consumer.commitSync();
                // 手动异步提交offset，当前线程提交offset不会阻塞，可以继续处理后面的程序逻辑
                /*consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + exception.getStackTrace());
                        }
                    }
                });*/
            }
        }
    }
}

Springboot整合

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

示例code

kafka-入门使用示例code.7z

启动脚本：
分布式，流

使用场景，依赖zk

概念

目前是2.4版本的

scala语言开发，依赖jvm
配置
设置
安全，不开启了，内部使用

api的使用

Java 进阶

kafka快速实战

1：官网：

2：场景：

3：基本概念

4：安装与使用

jvm环境

依赖zookeeper

下载

修改配置

启动服务

启动kafka，运行日志在logs目录的server.log文件里
bin/kafka‐server‐start.sh ‐daemon config/server.properties #后台启动，不会打印日志到控制台
或者用
bin/kafka‐server‐start.sh config/server.properties &

停止kafka
12 bin/kafka‐server‐stop.sh

server.properties核心配置

创建主题

发送消息

消费消息

单播消息

多播消息

查看消费组名：

主题topic和消息日志log

topic：

分区partition

创建多个分区的主题

查看topic情况

区别：

集群模式

配置文件

集群消费：

Producers

Consumer

消费顺序

JAVA客户端访问

消费组、

Springboot整合

示例code

kafka快速实战

1：官网：

2：场景：

3：基本概念

4：安装与使用

jvm环境

依赖zookeeper

下载

修改配置

启动服务

启动kafka，运行日志在logs目录的server.log文件里bin/kafka‐server‐start.sh ‐daemon config/server.properties #后台启动，不会打印日志到控制台或者用bin/kafka‐server‐start.sh config/server.properties &

停止kafka12 bin/kafka‐server‐stop.sh

server.properties核心配置

创建主题

发送消息

消费消息

单播消息

多播消息

查看消费组名：

主题topic和消息日志log

topic：

分区partition

创建多个分区的主题

查看topic情况

区别：

集群模式

配置文件

集群消费：

Producers

Consumer

消费顺序

JAVA客户端访问

消费组、

Springboot整合

示例code

启动kafka，运行日志在logs目录的server.log文件里
bin/kafka‐server‐start.sh ‐daemon config/server.properties #后台启动，不会打印日志到控制台
或者用
bin/kafka‐server‐start.sh config/server.properties &

停止kafka
12 bin/kafka‐server‐stop.sh