2，如果consumer group中的consumer线程数量比partition少，那么有的线程将会收到多个消息。并且不保证数据间的顺序性，kafka只保证在一个partition上数据是有序的，

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

4，High-level接口中获取不到数据的时候是会block的

关于consumer group（high api）的几点总结：
1，以consumer group为单位订阅 topic，每个consumer一起去消费一个topic；
2，consumer group 通过zookeeper来消费kafka集群中的消息（这个过程由zookeeper进行管理）；
相对于low api自己管理offset，high api把offset的管理交给了zookeeper，但是high api并不是消费一次就在zookeeper中更新一次，而是每间隔一个(默认1000ms)时间更新一次offset，可能在重启消费者时拿到重复的消息。此外，当分区leader发生变更时也可能拿到重复的消息。因此在关闭消费者时最好等待一定时间（10s）然后再shutdown。
3，consumer group 设计的目的之一也是为了应用多线程同时去消费一个topic中的数据。

例子：

1. import kafka.consumer.ConsumerIterator;
2. import kafka.consumer.KafkaStream;
4. public class ConsumerTest implements Runnable {
5. private KafkaStream m_stream;
6. private int m_threadNumber;
8. public ConsumerTest(KafkaStream a_stream, int a_threadNumber) {
9. m_threadNumber = a_threadNumber;
10. m_stream = a_stream;
11. }
13. public void run() {
14. ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
15. while (it.hasNext())
16. System.out.println(“Thread “ + m_threadNumber + “: “ + new String(it.next().message()));
17. System.out.println(“Shutting down Thread: “ + m_threadNumber);
18. }
19. }
21. //配置连接zookeeper的信息
22. private static ConsumerConfig createConsumerConfig(String a_zookeeper, String a_groupId) {
23. Properties props = new Properties();
24. props.put(“zookeeper.connect”, a_zookeeper); //zookeeper连接地址
25. props.put(“group.id”, a_groupId); //consumer group的id
26. props.put(“zookeeper.session.timeout.ms”, “400”);
27. props.put(“zookeeper.sync.time.ms”, “200”);
28. props.put(“auto.commit.interval.ms”, “1000”);
29. return new ConsumerConfig(props);
30. }
32. //建立一个消费者线程池
33. public void run(int a_numThreads) {
34. Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
35. topicCountMap.put(topic, new Integer(a_numThreads));
36. Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
37. List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
40. // now launch all the threads
41. //
42. executor = Executors.newFixedThreadPool(a_numThreads);
44. // now create an object to consume the messages
45. //
46. int threadNumber = 0;
47. for (final KafkaStream stream : streams) {
48. executor.submit(new ConsumerTest(stream, threadNumber));
49. threadNumber++;
50. }
51. }
53. //经过一段时间后关闭
54. try {
55. Thread.sleep(10000);
56. } catch (InterruptedException ie) {
58. }
59. example.shutdown();