（1）Customer和Customer Group
（1）两种常用的消息模型
队列模型（queuing）和发布-订阅模型（publish-subscribe）。
队列的处理方式是一组消费者从服务器读取消息，一条消息只由其中的一个消费者来处理。
发布-订阅模型中，消息被广播给所有的消费者，接收到消息的消费者都可以处理此消息。
（2）Kafka的消费者和消费者组
Kafka为这两种模型提供了单一的消费者抽象模型： 消费者组 （consumer group）。 消费者用一个消费者组名标记自己。 一个发布在Topic上消息被分发给此消费者组中的一个消费者。 假如所有的消费者都在一个组中，那么这就变成了队列模型。 假如所有的消费者都在不同的组中，那么就完全变成了发布-订阅模型。 一个消费者组中消费者订阅同一个Topic，每个消费者接受Topic的一部分分区的消息，从而实现对消费者的横向扩展，对消息进行分流。
注意：当单个消费者无法跟上数据生成的速度，就可以增加更多的消费者分担负载，每个消费者只处理部分partition的消息，从而实现单个应用程序的横向伸缩。但是不要让消费者的数量多于partition的数量，此时多余的消费者会空闲。此外，Kafka还允许多个应用程序从同一个Topic读取所有的消息，此时只要保证每个应用程序有自己的消费者组即可。
消费者组的概念就是：当有多个应用程序都需要从Kafka获取消息时，让每个app对应一个消费者组，从而使每个应用程序都能获取一个或多个Topic的全部消息；在每个消费者组中，往消费者组中添加消费者来伸缩读取能力和处理能力，消费者组中的每个消费者只处理每个Topic的一部分的消息，每个消费者对应一个线程。
（3）线程安全
在同一个群组中，无法让一个线程运行多个消费者，也无法让多线线程安全地共享一个消费者。按照规则，一个消费者使用一个线程，如果要在同一个消费者组中运行多个消费者，需要让每个消费者运行在自己的线程中。最好把消费者的逻辑封装在自己的对象中，然后使用java的ExecutorService启动多个线程，使每个消费者运行在自己的线程上,可参考https://www.confluent.io/blog
（2）Partition Rebalance分区再均衡（1）消费者组中新添加消费者读取到原本是其他消费者读取的消息
（2）消费者关闭或崩溃之后离开群组，原本由他读取的partition将由群组里其他消费者读取
（3）当向一个Topic添加新的partition，会发生partition在消费者中的重新分配
以上三种现象会使partition的所有权在消费者之间转移，这样的行为叫作再均衡。
再均衡的优点：
给消费者组带来了高可用性和伸缩性
再均衡的缺点：
（1）再均衡期间消费者无法读取消息，整个群组有一小段时间不可用
（2）partition被重新分配给一个消费者时，消费者当前的读取状态会丢失，有可能还需要去刷新缓存，在它重新恢复状态之前会拖慢应用程序。
因此需要进行安全的再均衡和避免不必要的再均衡。
（3）创建Kafka消费者、订阅主题、轮询Properties props = new Properties(); props.put("bootstrap", "broker1:9092,broker2:9092"); props.put("group.id", "CountryCounter"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); //1.创建消费者 KafkaConsuner<String, String> consumer = new KafkaConsumer<String, String>(props); //2.订阅Topic //创建一个只包含单个元素的列表，Topic的名字叫作customerCountries consumer.subscribe(Collections.singletonList("customerCountries")); //支持正则表达式，订阅所有与test相关的Topic //consumer.subscribe("test.*"); //3.轮询 //消息轮询是消费者的核心API，通过一个简单的轮询向服务器请求数据，一旦消费者订阅了Topic，轮询就会处理所欲的细节，包括群组协调、partition再均衡、发送心跳 //以及获取数据，开发者只要处理从partition返回的数据即可。 try { while (true) {//消费者是一个长期运行的程序，通过持续轮询向Kafka请求数据。在其他线程中调用consumer.wakeup()可以退出循环 //在100ms内等待Kafka的broker返回数据.超市参数指定poll在多久之后可以返回，不管有没有可用的数据都要返回 ConsumerRecord<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { log.debug(record.topic() + record.partition() + record.offset() + record.key() + record.value()); //统计各个地区的客户数量，即模拟对消息的处理 int updatedCount = 1; updatedCount += custCountryMap.getOrDefault(record.value(), 0) + 1; custCountryMap.put(record.value(), updatedCount); //真实场景中，结果一般会被保存到数据存储系统中 JSONObject json = new JSONObject(custCountryMap); System.out.println(json.toString(4)); } } } finally { //退出应用程序前使用close方法关闭消费者，网络连接和socket也会随之关闭，并立即触发一次再均衡 consumer.close(); }