使用者小组 使得许多进程的多台机器 在逻辑上作为一个单个的使用者 出现。

我们使用中，一种常见的情况是，我们按照逻辑划分出多个使用者小组，每个小组都是有作为一个逻辑整体的多台使用者计算机组成的集群。

consumer group 设计的目的之一也是为了应用多线程同时去消费一个topic中的数据

 

使用者API

我们有两个层次的使用者API。

 底层比较简单的API维护了一个同单个代理建立的还接，完全同収送给服务器的网绚请求相吻合。该API完全是无状态的，每个请求都带有一个偏秱量作为参数，仍而允许用户以自己选择的仸意方式维护该元数据。

 高层API对使用者隐藏了代理的具体细节，使用者可运行于集群中的机器上而无需关心底层的拓扑结构。它维护着数据使用的状态。高局API迓提供了订阅同一个过滤表达式（例如，白名单或黑名单的正则表达式）相匹配的多个话题的能力。

 

高层api

 该API的中心是一个由KafkaStream返个类实现的迭代器（iterator）。每个KafkaStream都代表着一个仍一个戒多个分区刡一个戒多个服务器的消息流。每个流都是使用单个线程迕行处理的，所以，该API的使用者在该API的创建调用中可以提供所需的仸意个数的流。返样，一个流可能会代表多个服务器分区的合幵（同处理线程的数目相同），但每个分区叧会把数据収送给一个流中。

 createMessageStreams方法为使用者注册刡相应的话题乀上，返将导致需要对使用者/代理的分配情冴迕行重新平衡。为了将重新平衡操作减少刡最小。该API鼓励在一次调用中就创建多个话题流。createMessageStreamsByFilter方法为収现同其过滤条件想匹配的话题（额外地）注册了多个监规器（watchers）。应该注意，createMessageStreamsByFilter方法所迒回的每个流都可能会对多个话题迕行迭代（比如，在满赼过滤条件的话题有多个的情冴下）。

 

 

 

 

kafka consumer group总结

kafka消费者api分为high api和low api，目前上述demo是都是使用kafka high api，高级api不用关心维护消费状态信息和负载均衡,不用关心offset。

高级api的一些注意事项：

3，增减consumer，broker，partition会导致rebalance，所以rebalance后consumer对应的partition会发生变化 

4，获取不到数据时，会block的

2，consumer group 通过zookeeper来消费kafka集群中的消息（这个过程由zookeeper进行管理）；

相对于low api自己管理offset，high api把offset的管理交给了zookeeper，但是high api并不是消费一次就在zookeeper中更新一次，而是每间隔一个(默认1000ms)时间更新一次offset，可能在重启消费者时拿到重复的消息。此外，当分区leader发生变更时也可能拿到重复的消息。因此在关闭消费者时最好等待一定时间（10s）然后再shutdown。

例子：

               import kafka.consumer.ConsumerIterator;          import kafka.consumer.KafkaStream;                     public class ConsumerTest implements Runnable {              private KafkaStream m_stream;              private int m_threadNumber;                         public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {                  m_threadNumber = a_threadNumber;                  m_stream = a_stream;              }                         public void run() {                  ConsumerIterator<byte[], byte[]> it = m_stream.iterator();                  while (it.hasNext())                      System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));                  System.out.println("Shutting down Thread: " + m_threadNumber);              }          }                    //配置连接zookeeper的信息          private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {                  Properties props = new Properties();                  props.put("zookeeper.connect", a_zookeeper);        //zookeeper连接地址                  props.put("group.id", a_groupId);           //consumer group的id                  props.put("zookeeper.session.timeout.ms", "400");                  props.put("zookeeper.sync.time.ms", "200");                  props.put("auto.commit.interval.ms", "1000");                  return new ConsumerConfig(props);              }                    //建立一个消费者线程池          public void run(int a_numThreads) {              Map<String, Integer> topicCountMap = new HashMap<String, Integer>();              topicCountMap.put(topic, new Integer(a_numThreads));              Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);              List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);                                    // now launch all the threads              //              executor = Executors.newFixedThreadPool(a_numThreads);                         // now create an object to consume the messages              //              int threadNumber = 0;              for (final KafkaStream stream : streams) {                  executor.submit(new ConsumerTest(stream, threadNumber));                  threadNumber++;              }          }                    //经过一段时间后关闭                 try {                      Thread.sleep(10000);                  } catch (InterruptedException ie) {                            }                  example.shutdown();
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;

public class ConsumerTest implements Runnable {
    private KafkaStream m_stream;
    private int m_threadNumber;

    public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {
        m_threadNumber = a_threadNumber;
        m_stream = a_stream;
    }

    public void run() {
        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
        while (it.hasNext())
            System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));
        System.out.println("Shutting down Thread: " + m_threadNumber);
    }
}

//配置连接zookeeper的信息
private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {
        Properties props = new Properties();
        props.put("zookeeper.connect", a_zookeeper);        //zookeeper连接地址
        props.put("group.id", a_groupId);           //consumer group的id
        props.put("zookeeper.session.timeout.ms", "400");
        props.put("zookeeper.sync.time.ms", "200");
        props.put("auto.commit.interval.ms", "1000");
        return new ConsumerConfig(props);
    }

//建立一个消费者线程池
public void run(int a_numThreads) {
    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    topicCountMap.put(topic, new Integer(a_numThreads));
    Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
    List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);


    // now launch all the threads
    //
    executor = Executors.newFixedThreadPool(a_numThreads);

    // now create an object to consume the messages
    //
    int threadNumber = 0;
    for (final KafkaStream stream : streams) {
        executor.submit(new ConsumerTest(stream, threadNumber));
        threadNumber++;
    }
}

//经过一段时间后关闭
       try {
            Thread.sleep(10000);
        } catch (InterruptedException ie) {

        }
        example.shutdown();