中间件-Kafka_kafka中间件

消息队列的使用场景

• 流量缓冲；
• 系统解耦；
• 异构继承；
• 异步处理；
• 流处理；

消息队列的两种模式

• 点对点；
• 广播；

消息队列的产品

• Kafka
• RocketMQ
• ActiveQM
• RobbitMQ

Kafka的核心组件

Producer

• KeySerializer
• ValueSerializer
• Interceptor
• Partitioner
• RecordAccumulator
• Sender
• NetworkClient
• Selector

Broker

• Partition Leader
• Partition Follower
• Controller
• Coordinator
• Log
• Segment
• index
• timeindex
• AR/OR/ISR

Consumer

• ConsumerGroup
• Consumer
• ConsumerNetworkClient
• CompletedFetchQueue
• Interceptor
• KeyDeserializer
• ValueDeserializer

基础架构

• 为了提高Kafka的性能，分区工作在Producer端完成；
• 为了增加生产端的吞吐量，将一个topic分成多个partition；
• 为了保证数据安全，为每个partition引入一组follower进行数据的备份；
• 为了配合partition分区，消费端引入了消费组，组内每个消费组消费一部分partition数据；
• zookeeper中记录kafka集群的一些元数据；

集群安装

1.下载zookeeper安装包

​
wget https://dlcdn.apache.org/zookeeper/zookeeper-3.8.0/apache-zookeeper-3.8.0-bin.tar.gz

2.解压

​
tar -xzvf apache-zookeeper-3.8.0-bin.tar.gz

3.修改配置文件

mv zoo_simple.cfg zoo.cfg
vi zoo.cfg
dataDir=/Users/dongzhang/develop/apps/apache-zookeeper-3.8.0/datas

4.启动zookeeper

./zkServer.sh start

5.下载kafka安装包

wget https://archive.apache.org/dist/kafka/3.0.0/kafka_2.12-3.0.0.tgz

6.解压kafka安装包

tar -xzvf kafka_2.12-3.0.0.tgz

7.整理目录

cp -R kafka_2.12-3.0.0/  kafka_node1/ 
cp -R kafka_2.12-3.0.0/  kafka_node2/
cp -R kafka_2.12-3.0.0/  kafka_node3/
mv   kafka_2.12-3.0.0/ kafka_client

8.修改环境变量

vi ~/.bash_profile
 export KAFKA_HOME=/Users/dongzhang/develop/apps/kafka_cluster_3.0.0/kafka_client
 export PATH=$PATH:$KAFKA_HOME/bin
source ~/.bash_profile

9.修改server.properties

node1-3:
    broker.id=0~2
    listeners=PLAINTEXT://:9092、9093、9094
    log.dirs=/Users/dongzhang/develop/apps/kafka_cluster_3.0.0/kafka_node[1-3]/datas

10.启动kafka节点

kafka_node1/bin/kafka-server-start.sh -daemon kafka_node1/config/server.properties
kafka_node2/bin/kafka-server-start.sh -daemon kafka_node2/config/server.properties
kafka_node3/bin/kafka-server-start.sh -daemon kafka_node3/config/server.properties

11.测试

kafka-topics.sh --bootstrap-server localhost:9092 --list
#创建topic
kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 2 --replication-factor 3 --topic test
 
#启动生产者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test
#启动消费者
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topi
c test
#查看topic详情
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic test
Topic: test	TopicId: L_qX7_mtRHSP3jsuj079uQ	PartitionCount: 2	ReplicationFactor: 3	Configs: segment.bytes=1073741824
	Topic: test	Partition: 0	Leader: 1	Replicas: 1,0,2	Isr: 1,0,2
	Topic: test	Partition: 1	Leader: 0	Replicas: 0,2,1	Isr: 0,2,1

生产者

消息发送原理

在发送数据的过程中涉及到两个线程，Producer主线程和Sender线程，在主线程中会为每一个分区创建一个双端队列RecordAccumulator，主线程将消息放到RecordAccumulator中，然后Sender线程不断的从RecordAccumulator中拉取消息，发送到Broker。

分区策略

对topic进行分区有以下两个好处

• 便于合理使用存储资源，每个partition放到一个broker上，可以按照分区把topic中的数据切分成不同的部分进行存储，实现存储空间的负载均衡；
• 提高并行度，生产者可以以分区为单位发送数据，消费者可以以分区为单位存储数据；

配置参数

服务端

Zookeeper中的数据

主要有三个重要数据：brokers、partition、controller leader。 

Broker工作流程

Leader选举流程

1. broker在启动时，向zookeeper注册自己（/brokers/ids）;
2. 每一个broker的controller组件会以抢占的方式向zookeeper注册自己（/controller）,注册成功的controller作为leader来管理分区leader的选举;
3. controller监控/brokers/ids节点的变化，如果有节点加入或退出，就会进行后续处理；
4. 在isr列表中的节点都存活的前提下，controller按照ar列表的顺序来选举分区leader；
5. 选出分区leader后，将信息注册到zookeeper上（/brokers/topics/queue/paritition/0/state）；
6. 其他节点上的controller从zookeeper上同步leader选举结果；
7. 如果leader节点挂了，/brokers/ids节点内容就会发生变化，controller就会监控到；
8. controller从zookeeper获取到isr列表，并按照规则选出一个新的leader；

AR/OR/ISR

AR是所有的Partition副本id列表，OR是那些由于宕机或数据同步延迟过大被leader提出的partition的id列表，ISR是能够正常从partition leader同步数据的follower id，即AR=OR+ISR。

Leader选举规则

以在ISR列表中的节点是存活的为前提，按照AR列表中排在前面的优先。例如ISR列表为[0,1,2]，AR列表为[1,0,2]，那么久按照AR的顺序从ISR列表中选举新的partition作为Leader。

LEO与HW截断机制

• LEO：Log End Offset，LEO值 = 每个分区的最后一个offset + 1；
• HW：Hight Watermark，所有分区副本中最小的LEO；

由于分区leader和follower之间存在数据延迟，如果leader节点出现了问题，并且正好有一部分数据还没有被follower完整同步，那么当controller选出新的leader时，剩余的各个follower可能就会出现不一致的情况，为了保证这种极端情况下数据的一致性，新的leader就会以HW为基准，把某些节点中多余的数据给删除掉，虽然会造成数据丢失，但能够保证数据的一致性。

从kafka 服务端的角度看会出现数据丢失，但在生产端如果我们把acks设置为-1，那么当出现上述情况时，就会到时producer发送消息失败，这样就会做后续的重发补偿等处理，保证数据不丢，但如果acks=0或1，就会出现数据丢失的情况。

Partition Leader的负载均衡

正常情况下，Kafka本身就会把分区Leader均匀的分布到各个机器上，来保证每台机器的读写吞吐量基本均匀，但如果某些broker宕机，会导致某些partition leader过于集中到某些机器上，这会导致少数几台broker读写压力过大，其他宕机的broker重启之后都是作为partition follower存在的，读写压力比较低，这就造成了集群的负载不均衡。

Kafka提供了参数来Partition Leader自平衡机制

默认分区策略

• 如果发送消息时制定了分区号，就将消息发送到制定的分区；
• 如果没有指定分区号但指定了key，则按照key.hashCode() % 分区数 选择分区；
• 如果没有指定分区或分区，则当前批次的数据会随机发往同一个分区，下一个批次再重新选择分区号；

文件存储

文件清理 

• 删除
• 压缩

Kafka中的日志数据默认保持7天，可以通过调整以下参数来调整

高效读写原理

• 利用分区提高并行度；
• 利用稀疏索引，提高数据检索效率；
• 顺序写磁盘；
• 页缓存和零拷贝；

配置参数

消费者

消费模式

Consumer工作流程

1. 通过ConsumerNetworkClient向某个broker上的partition发送拉取数据的请求；
2. broker返回一批数据，这些数据被放到一个队列当中；
3. 最终返回到调用处，进行反序列化齐、拦截器的处理后将原始数据返回，并进行业务逻辑的处理；

消费者组原理

消费者组由多个consumer组成，形成一个消费者组的条件是所有消费者的group id相同。

• 消费者组内每个消费者负责消费不同分区内的数据，一个分区只能由消费者组内的一个消费者消费；
• 消费者组之间互不影响，所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者；
• 如果消费者组中的消费者数量多于topic的分区数量，则会有空闲的consumer；

初始化流程 

消费者组在初始化的时候需要Coordinator这个组件协助，在每一个Broker上都有Coordinator这个组件，消费者组在初始化的时候会根据gourpId.hashCode() % 50来选择某一个Coordinator，这里的50实际上是__consumer_offset这个topic的分区数量，比如groupId的hashCode=1，1%50=1，那么__consumer_offset这个topic的1号分区在哪一个broker上就选择这个broker上的Coordinator组件进行协调，具体初始化流程如下：

1. 组内的每一个consumer都向指定的Coordinator发送JoinGroup请求；
2. Coordinator会从所有的Consumer中选出一个作为Leader；
3. Coordinator把待消费的topic的信息发送给这个Consumer Leader；
4. Consumer Leader根据具体的规则来制定消费分区方案；
5. 确定完成后把消费方案发给Coordinator；
6. Coordinator再将这个方案发给其他的Consumer；
7. 每个Consumer都会与Coordinator保持心跳，默认频率为3s，一旦超时（session.timeout.ms=45s）该消费者就会被剔除，并出发再平衡；如果消费者处理消息的时间过长（max.poll.interval.ms=5m）也会触发再平衡；

分区分配再平衡

当有多个消费者去消费多个分区的时候，在Coordinator的协助下，会将不同的partition分配给不同的消费者去处理，Kafka提供了以下几种分配策略：

1. Range
2. RandRobin
3. Sticky
4. CooperativeSticky

可以通过参数partition.assignment.strategy来修改，而当某个消费者发生故障被Coordinator剔除或有新的消费者加入时，就会触发再平衡，也就是在这些消费者之间充分分配分区。

Range策略首先会对topic内的分区进行编号并排序，并对消费者按字母顺序排序，比如有7个分区和3个消费者，那么分区编号就是0~6，消费者编号就是C0~C2，然后通过 分区数/消费者数 来决定每个消费者应该消费几个分区，如果除不尽，那么前面几个消费者会多分到1个分区。这种分配方式的弊端在于，如果只针对一个topic而言某个消费者多消费一个分区的数据影响还不是很大，但如果有多个topic，那么针对每个topic 排在前面的消费者都会多消费分区，这样就会导致Consumer的负载不均衡。

RoundRobin是轮询策略，是把所有parition和所有consumer都列出来，然后按照hashCode进行排序，最后通过轮询算法给各个consumer分配partition。

RoundRobin是轮询策略，是把所有parition和所有consumer都列出来，然后按照hashCode进行排序，最后通过轮询算法给各个consumer分配partition。

Sticky可以理解为分配的结果带有一定的粘性，即在执行一次新的分配之前，会考虑上一次的分配结果，在同一消费者组内如果消费者出现问题，会尽量保持原有的分区分配不发生变化。

//修改分区分配策略
//org.apache.kafka.clients.consumer.RangeAssignor
//org.apache.kafka.clients.consumer.RoundRobinAssignor
//org.apache.kafka.clients.consumer.StickyAssignor
//org.apache.kafka.clients.consumer.CooperativeStickyAssignor
List<String> strateies = new ArrayList<>();
strateies.add("org.apache.kafka.clients.consumer.RoundRobinAssignor");
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, strateies);

维护Offset

在kafka 0.9之前，消费端消费数据的offset维护在zookeeper上，但由于存在性能问题，从0.9版本以后就将offset存放在了kafka的一个指定的topic中(__consumer_offset)，某个consumer如果想确认或提交自己的offset就去消费这个topic，内部消息的key=groupId + topic + 分区号，value=offset，卡夫卡内部每隔一段时间就是对这个topic执行compact，也就是每个值保留最新的条数据，即最新提交的offset值。

默认情况下，消费者自动维护offset的提交，但可以通过以下两个参数对提交offset的行为进行调整：

如果enable.auto.commit=false，就需要手动提交事务，kafka提供了手动同步提交（commitSync）和手动异步提交（commitAsync）两种方式，两者的相同点是都会将本地数据的最大偏移量提交到__consumer_offset，不同点是同步提交会阻塞当前线程一直到提交成功为止，并且会自动进行失败重试，而异步提交方式没有失败重试机制，并且也不会阻塞当前线程的执行，所以可能会导致提交offset失败。

• commitSync：必须等待提交完成才能消费下一批数据；
• commitAsync：发送完成提交offset的请求后就开始消费下一批数据；

同时，Kafka提供了三种offset的消费模式：earlist | least | none

• earlist是从最早的offset进行消费；
• least是从最新的偏移量开始消费；
• none是如果找不到消费者组之前的消费偏移量就会抛异常；

幂等性原理

幂等性就是指不论Producer向Broker发送多少条重复属于，Consumer只会收到一条，保证不会重复消费，所以精准一次=幂等性 + 至少一次，Broker对数据进行重复判断的标准是相同时，就只会持久化一条数据，其中PID是Producer每次重启时都会重新分配，Partition为分区号，SeqNumber是单调递增的，所以，幂等性只能保证单分区单会话内不重复。

配置参数

实践经验

生产端

如何提高生产端的吞吐量

1. 增加topic分区数；
2. 增加缓冲区大小；
3. 增加等待时间；
4. 增加批次大小；
5. 开启消息压缩（仅限于大文本消息）；

如何保证数据的可靠性（不丢数据）

1. 为了保证让消息正确的发送到服务端，需要将acks设置为-1
2. 由于存在LEO和HW截断机制，需要保证可用partition follower的数量（即ISR列表）要 >= 2；
3. 做好重试参数retries的配置（默认为Integer的最大值），以及消费端发送失败的处理（记录日志或消息补偿）；

如何实现最多一次消费

• ack=0，即发送出去的消息不进行确认，但可能会丢消息，一般用在处理一些不太重要的数据，比如日志收集；

如何实现最少一次消费

1. 保证partition follower(ISR列表) >= 2;
2. acks=-1；
3. 因为网络等外部原因，可能会因为重试而造成多次发送，最终导致多次消费；

如何实现精准一次

• 开启幂等性；
• 发送端用事务api；

如何保证数据的有序性

1. 单分区
2. max.in.flight.requests.pre.connections=1（Sender内部有缓存，当某个批次发送失败时触发的重试可能会导致乱序），如果在kafka1.x以后，开启了幂等性，那么只需要将max.in.flight.requests.pre.connections设置为<=5即可，因为kafka服务端会缓存producer发来的最近5个request的元数据，故无论如何，都可以保证最近5个request的数据都是有序的。

服务端

手动调整分区副本存储分布

1. 创建一个json文件 increase-replication-factor.json
2. 指定副本分布配置（replicas）

3. {
      "version":1,
      "partitions":[
           {"topic":"three","partition":0,"replicas":[0,1]},
           {"topic":"three","partition":1,"replicas":[0,1]},
           {"topic":"three","partition":2,"replicas":[1,0]},
           {"topic":"three","partition":3,"replicas":[1,0]}
       ] 
   }
   

4. bin/kafka-reassign-partitions.sh -- bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --execute

5. bin/kafka-reassign-partitions.sh -- bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --verify

6. bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic three

增加副本因子

1. 创建文件 increase-replication-factor.json

2. {
       "version":1,
       "partitions":[
           {"topic":"four","partition":0,"replicas":[0,1,2]},            
           {"topic":"four","partition":1,"replicas":[0,1,2]},
           {"topic":"four","partition":2,"replicas":[0,1,2]}
       ]
   }
   

3. bin/kafka-reassign-partitions.sh -- bootstrap-server localhost:9092 --reassignment-json-file increase-replication-factor.json --execute

消费端

消费者事务

如果想完成Consumer端的精准一次性消费，那么需要Kafka消费端将消费过程和提交offset 过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质，比如Mysql的事务。

处理消息挤压

1. 增加消费者数量（数量没有达到分区数时）；
2. 增加消费端拉取一个批次的数据的大小（fetch.max.bytes，默认时50m）；
3. 增加一次拉取消息的条数（max.poll.records，默认时500条）；

但要根据下游的处理能力来定，如果超过了下游的处理能力，那么即使增加Kafka单批次的拉取量，也无法快速处理掉。

代码示例

发送端

同步发送

/**
 * 同步发送
 * 主线程会等待Sender成功将消息发送到partition 并收到ack后才继续发送下一个
 * Callback会在Sender收到partition的ack后 回调
 * */
public class SyncProducer {
    
    public void asyncWithoutCallback() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 5;i ++) {
            producer.send(new ProducerRecord<>("test", "test-" + i));
        }
        producer.close();
    }
 
 
    public void asyncWithCallback() throws Exception {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 5000;i ++) {
            producer.send(new ProducerRecord<>("queue", "test-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append(recordMetadata.topic() + " ");
                    stringBuilder.append(recordMetadata.partition() + " ");
                    stringBuilder.append(recordMetadata.offset() + " ");
                    stringBuilder.append(recordMetadata.timestamp() + " ");
                    System.out.println(stringBuilder.toString());
                }
            }).get();
        }
        producer.close();
    }
 
    public static void main(String[] args) throws Exception {
        SyncProducer producer = new SyncProducer();
        //producer.asyncWithoutCallback();
        producer.asyncWithCallback();
    }
}

异步发送

/**
 * 异步发送
 * 在异步发送模式下，主线程将消息放入到RecordAccumulator中就返回了，不会等待
 * Sender把数据成功发送到Partition中
 *
 * 而Callback会在Sender收到partition的ack后 回调
 * */
public class AsyncProducer {
 
 
 
    public void asyncWithoutCallback() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 5000;i ++) {
            producer.send(new ProducerRecord<>("test", "test-" + i));
        }
        producer.close();
    }
 
 
    public void asyncWithCallback() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 5000;i ++) {
            producer.send(new ProducerRecord<>("test", "test-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append(recordMetadata.topic() + " ");
                    stringBuilder.append(recordMetadata.partition() + " ");
                    stringBuilder.append(recordMetadata.offset() + " ");
                    stringBuilder.append(recordMetadata.timestamp() + " ");
                    System.out.println(stringBuilder.toString());
                }
            });
        }
        producer.close();
    }
 
    public static void main(String[] args) {
        AsyncProducer producer = new AsyncProducer();
        //producer.asyncWithoutCallback();
        producer.asyncWithCallback();
    }
}

指定分区号/指定key/自定义分区器

/**
 * 将消息发送到指定的分区.
 * */
public class SpecificPartitionProducer {
 
    public void sendToSpecificPartition() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 5000;i ++) {
            //将消息发送到0号分区
            producer.send(new ProducerRecord<>("test", 0, "", "test-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append(recordMetadata.topic() + " ");
                    stringBuilder.append(recordMetadata.partition() + " ");
                    stringBuilder.append(recordMetadata.offset() + " ");
                    stringBuilder.append(recordMetadata.timestamp() + " ");
                    System.out.println(stringBuilder.toString());
                }
            });
        }
        producer.close();
    }
 
    public static void main(String[] args) {
        SpecificPartitionProducer producer = new SpecificPartitionProducer();
        producer.sendToSpecificPartition();
    }
}
 
 
 
/**
 * 将消息发送到指定的分区.
 * */
public class SpecificKeyProducer {
 
    public void sendToSpecificPartition() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 5;i ++) {
            //将消息发送到0号分区
            //分别测试  a - 0 / b - 0 / c - 0 / f - 1
            producer.send(new ProducerRecord<>("test", "f", "test-" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append(recordMetadata.topic() + " ");
                    stringBuilder.append(recordMetadata.partition() + " ");
                    stringBuilder.append(recordMetadata.offset() + " ");
                    stringBuilder.append(recordMetadata.timestamp() + " ");
                    System.out.println(stringBuilder.toString());
                }
            });
        }
        producer.close();
    }
 
    public static void main(String[] args) {
        SpecificKeyProducer producer = new SpecificKeyProducer();
        producer.sendToSpecificPartition();
    }
}
 
 
 
/**
 * 自定义分区规则
 * */
public class CustomPartitioner implements Partitioner {
 
    /**
     *
     * */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        String msgValue = value.toString();
        int partition;
        if (msgValue.contains("kafka")) {
            partition = 1;
        } else {
            partition = 0;
        }
        return partition;
    }
 
    @Override
    public void close() {
 
    }
 
    @Override
    public void configure(Map<String, ?> map) {
 
    }
}
 
 
/**
 * 将消息发送到指定的分区.
 * */
public class SpecificCustomPartitionerProducer {
 
    public void sendToSpecificPartition() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, CustomPartitioner.class.getName());
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 500;i ++) {
            //将消息发送到0号分区
            //分别测试  a - 0 / b - 0 / c - 0 / f - 1
            String value = "test-";
            if (i % 2 == 0) {
                value = value + "kafka";
            }else {
                value = value + "kafka";
            }
            //producer.send(new ProducerRecord<>("queue", 0, "", value), new Callback() {
            producer.send(new ProducerRecord<>("queue", value), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append(recordMetadata.topic() + " ");
                    stringBuilder.append(recordMetadata.partition() + " ");
                    stringBuilder.append(recordMetadata.offset() + " ");
                    stringBuilder.append(recordMetadata.timestamp() + " ");
                    System.out.println(stringBuilder.toString());
                }
            });
        }
        producer.close();
    }
 
    public static void main(String[] args) {
        SpecificCustomPartitionerProducer producer = new SpecificCustomPartitionerProducer();
        producer.sendToSpecificPartition();
    }
}

优化后的高吞吐生产者

/**
 * 优化生产端的吞吐量.
 * */
public class OptimizeProducer {
 
    public void send() throws Exception {
 
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig. VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        //默认时16k，够用了
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        //等待时间，默认时0
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 50);
        //缓冲区大小，默认时32MB，调整到64MB
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);
        //设置压缩类型
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 100;i ++) {
            String bigMsg = i + "日常生活: 手机号码查询 2021年放假安排 邮编查询 长途电话区号 货币汇率查询 家常菜谱大全 人民币存款利率表 下载地址转换 北京时间 大学查询 汽车车标大全 快递查询 国家地区查询 升降旗时间 常用电话号码 电费计算器 日期差计算 网速测试 数字大写转换 今日油价 个税计算器 国际天气预报 亲属关系计算 台湾邮编查询 (共31个) 占卜求签: 指纹运势查询 生男生女预测 预测吉凶 称骨算命 黄大仙灵签 六十四卦金钱课 观音灵签 诸葛神算 妈祖天后灵签 关帝灵签 吕祖灵签 车公灵签 王公祖仔灵签 月老灵签 文王神卦 灵棋经 二十八星宿算命 佛祖灵签 月老姻缘签 (共20个) 民俗文化: 老黄历 万年历 周公解梦大全 十二生肖 歇后语大全 百家姓 民间谚语 二十四节气表 历史朝代表 解密生日 名人名言名句大全 古兰经 基督教圣经 三字经 地母经 (共16个) 交通出行: 列车时刻表 全国各地车牌号查询 车辆违章查询 北京时间校准 机场三字码查询 实时交通路况 地铁线路图 车牌限行查询 火车票代售点 中国电子地图 交通标志 (共11个) 学习应用: 汉语字典 汉语词典 成语大全 在线翻译 计算器 在线输入法 圆周率 繁体字转换器 汉字拼音查询 摩尔斯电码 存储单位换算器 时间换算器 英文名 长度换算器 温度换算器 重量换算器 体积换算器 功率换算器 面积换算器 压力换算器 热量换算器 五笔字根表 区位码查询 笔画数查询 汉字部首查询 郑码编码查询 仓颉编码查询 中文电码查询 四角号码查询 英文缩写大全 组词大全 近义词 (共32个) 休闲娱乐: 号码吉凶预测 脑筋急转弯 中华谜语大全 竖排古文 火星文转换 QQ价值评估 QQ在线状态查询 外星体重 外星年龄 在线拆字 笑话大全 绕口令大全 (共15个) 站长工具: IP地址查询 密码强度检测 时间戳转换 ASCII码对照表 HTML/JS互转 BASE64加密解密 MD5加密解密 进程查询 在线编码解码 谷歌PR查询 搜狗SR查询 网站速度测试 二维码生成器 颜色代码表 HTML特殊符号 CSS在线解压缩 JS在线解压缩 HTML代码调试器 密码生成器 (共20个) 身体健康: 安全期计算器 预产期计算器 体质指数计算器 食物营养价值 粥谱大全 生星座宝宝 身高预测 血型与性格 ";
            producer.send(new ProducerRecord<String, String>("test", bigMsg), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append(recordMetadata.topic() + " ");
                    stringBuilder.append(recordMetadata.partition() + " ");
                    stringBuilder.append(recordMetadata.offset() + " ");
                    stringBuilder.append(recordMetadata.timestamp() + " ");
                    System.out.println(stringBuilder.toString());
                }
            });
        }
 
        //如果发送的消息比较少，就需要sleep，否则，因为sender会等待50ms，但那时producer线程就
        //退出了
        TimeUnit.SECONDS.sleep(3);
    }
 
    public static void main(String[] args) throws Exception {
        OptimizeProducer producer = new OptimizeProducer();
        producer.send();
    }
}

保证数据可靠性

/**
 * 保证数据可靠性的Producer
 * acks=-1
 * partition num >= 3
 * */
public class DataReliabilityProducer {
 
    public void send() throws Exception {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        //高可用设置
        properties.put(ProducerConfig.ACKS_CONFIG, "-1");
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
        for (int i = 0;i < 100;i ++) {
            String bigMsg = "message-";
            producer.send(new ProducerRecord<String, String>("test", bigMsg + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    StringBuilder stringBuilder = new StringBuilder();
                    stringBuilder.append(recordMetadata.topic() + " ");
                    stringBuilder.append(recordMetadata.partition() + " ");
                    stringBuilder.append(recordMetadata.offset() + " ");
                    stringBuilder.append(recordMetadata.timestamp() + " ");
                    System.out.println(stringBuilder.toString());
                }
            });
        }
 
        TimeUnit.SECONDS.sleep(3);
    }
 
    public static void main(String[] args) throws Exception {
        DataReliabilityProducer producer = new DataReliabilityProducer();
        producer.send();
    }
}

精准发送一次

/**
 * 精准发送一次
 * */
public class OnlyOnceProducer {
 
    public void send() {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        properties.put(ProducerConfig. VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
 
        properties.put(ProducerConfig.ACKS_CONFIG, "-1");
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_0");
 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(properties);
 
        //初始化事务
        producer.initTransactions();
        //开启事务
        producer.beginTransaction();
        try {
            for (int i = 0; i < 100; i++) {
                producer.send(new ProducerRecord<>("test", "test--" + i));
            }
            //int i = 10 / 0;
            //提交事务
            producer.commitTransaction();
        }catch (Exception ex) {
            System.out.println("发送异常，终止事务");
            //终止事务
            producer.abortTransaction();
        } finally {
            producer.close();
        }
 
    }
 
    public static void main(String[] args) {
        OnlyOnceProducer producer = new OnlyOnceProducer();
        producer.send();
    }
}

消费端

一个普通的消费者

/**
 * 独立消费者.
 * */
public class SingleConsumer {
 
    public void pull() {
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "single_consumer_group");
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);
 
        //拉取数据
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
        }
    }
 
    public static void main(String[] args) {
        SingleConsumer consumer = new SingleConsumer();
        consumer.pull();
    }
}

消费者组

public class Consumer1 {
 
    public void pull() {
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "single_consumer_group");
 
        //修改分区分配策略
        //org.apache.kafka.clients.consumer.RangeAssignor
        //org.apache.kafka.clients.consumer.RoundRobinAssignor
        //org.apache.kafka.clients.consumer.StickyAssignor
        //org.apache.kafka.clients.consumer.CooperativeStickyAssignor
        List<String> strateies = new ArrayList<>();
        strateies.add("org.apache.kafka.clients.consumer.RoundRobinAssignor");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, strateies);
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);
 
        //拉取数据
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
        }
    }
 
    public static void main(String[] args) {
        Consumer1 consumer = new Consumer1();
        consumer.pull();
    }
}
 
 
 
public class Consumer2 {
 
    public void pull() {
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "single_consumer_group");
 
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);
 
        //拉取数据
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
        }
    }
 
    public static void main(String[] args) {
        Consumer2 consumer = new Consumer2();
        consumer.pull();
    }
}
 
 
 
public class Consumer3 {
 
    public void pull() {
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "single_consumer_group");
 
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);
 
        //拉取数据
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
        }
    }
 
    public static void main(String[] args) {
        Consumer3 consumer = new Consumer3();
        consumer.pull();
    }
}

手动提交offset

/**
 * 手动提交offset.
 * @author dongzhang
 * */
public class ManualCommitConsumer {
 
    public void run() {
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "manual_commit_group");
 
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);
 
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
 
            //同步提交
            consumer.commitAsync();
            //异步提交
            //consumer.commitAsync();
        }
    }
 
    public static void main(String[] args) {
        ManualCommitConsumer consumer = new ManualCommitConsumer();
        consumer.run();
    }
}

指定offset消费

/**
 * 从任意指定的偏移量进行消费.
 * @author dongzhang
 * */
public class ManualOffsetConsumer {
 
    public void run() {
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "single_consumer_group");
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);
 
        //拿到当前的分区配置
        Set<TopicPartition> set = new HashSet<>();
        while (set.isEmpty()) {
            consumer.poll(Duration.ofSeconds(1));
            set = consumer.assignment();
        }
 
        //遍历所有的分区，并指定
        for (TopicPartition topicPartition : set) {
            consumer.seek(topicPartition, 1000);
        }
 
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : records) {
                System.out.println(consumerRecord);
            }
        }
    }
 
    public static void main(String[] args) {
        ManualOffsetConsumer consumer = new ManualOffsetConsumer();
        consumer.run();
    }
}

指定时间戳消费

/**
 * 从指定的时间开始下消费
 * */
public class ManualTimeConsumer {
 
    public void run() {
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "single_consumer_group");
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);
 
        //拿到当前的分区配置
        Set<TopicPartition> set = new HashSet<>();
        while (set.isEmpty()) {
            consumer.poll(Duration.ofSeconds(1));
            set = consumer.assignment();
        }
 
        HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>();
        // 封装集合存储，每个分区对应一天前的数据
        for (TopicPartition topicPartition : set) {
            timestampToSearch.put(topicPartition,
                    System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
        }
 
        // 获取从 1 天前开始消费的每个分区的 offset
        Map<TopicPartition, OffsetAndTimestamp> offsets = consumer.offsetsForTimes(timestampToSearch);
        // 遍历每个分区，对每个分区设置消费时间。
        for (TopicPartition topicPartition : set) {
            OffsetAndTimestamp offsetAndTimestamp = offsets.get(topicPartition);
            // 根据时间指定开始消费的位置
            if (offsetAndTimestamp != null){
                consumer.seek(topicPartition, offsetAndTimestamp.offset());
            }
        }
 
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : records) {
                System.out.println(consumerRecord);
            }
        }
    }
}

消费指定的分区

/**
 * 独立消费者.
 * 消费数据时，指定某个分区
 * */
public class SingleAndSpecificPartitionConsumer {
 
    public void pull() {
 
        //配置属性
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "single_consumer_group");
 
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
        /*List<String> topics = Arrays.asList("queue");
        consumer.subscribe(topics);*/
 
        //指定从queue topic的第0个分区消费数据
        List<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition("queue", 0));
        consumer.assign(topicPartitions);
 
        //拉取数据
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println(record);
            }
        }
    }
 
    public static void main(String[] args) {
        SingleAndSpecificPartitionConsumer consumer = new SingleAndSpecificPartitionConsumer();
        consumer.pull();
    }
}

优化建议