使用默认接入点收发消息

   import java.util.ArrayList;
   import java.util.List;
   import java.util.Properties;

   import org.apache.kafka.clients.consumer.ConsumerConfig;
   import org.apache.kafka.clients.consumer.ConsumerRecord;
   import org.apache.kafka.clients.consumer.ConsumerRecords;
   import org.apache.kafka.clients.consumer.KafkaConsumer;
   import org.apache.kafka.clients.producer.ProducerConfig;

   public class KafkaConsumerDemo {
       public static void main(String args[]) {
           //加载kafka.properties。
           Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

           Properties props = new Properties();
           //设置接入点，请通过控制台获取对应Topic的接入点。
           props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
           //两次Poll之间的最大允许间隔。
           //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Group移除并触发Rebalance，默认30s。
           props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
           //每次Poll的最大数量。
           //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
           props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
           //消息的反序列化方式。
           props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
           props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
           //当前消费实例所属的消费组，请在控制台申请之后填写。
           //属于同一个组的消费实例，会负载消费消息。
           props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
           //构造消费对象，也即生成一个消费实例。
           KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
           //设置消费组订阅的Topic，可以订阅多个。
           //如果GROUP_ID_CONFIG是一样，则订阅的Topic也建议设置成一样。
           List<String> subscribedTopics =  new ArrayList<String>();
           //如果需要订阅多个Topic，则在这里添加进去即可。
           //每个Topic需要先在控制台进行创建。
           String topicStr = kafkaProperties.getProperty("topic");
           String[] topics = topicStr.split(",");
           for (String topic: topics) {
               subscribedTopics.add(topic.trim());
           }
           consumer.subscribe(subscribedTopics);

           //循环消费消息。
           while (true){
               try {
                   ConsumerRecords<String, String> records = consumer.poll(1000);
                   //必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。
                   //建议开一个单独的线程池来消费消息，然后异步返回结果。
                   for (ConsumerRecord<String, String> record : records) {
                       System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
                   }
               } catch (Exception e) {
                   try {
                       Thread.sleep(1000);
                   } catch (Throwable ignore) {

                   }
                   e.printStackTrace();
               }
           }
       }
   }

   import java.util.ArrayList;
   import java.util.List;
   import java.util.Properties;
   import java.util.concurrent.atomic.AtomicBoolean;
   import org.apache.kafka.clients.consumer.ConsumerConfig;
   import org.apache.kafka.clients.consumer.ConsumerRecord;
   import org.apache.kafka.clients.consumer.ConsumerRecords;
   import org.apache.kafka.clients.consumer.KafkaConsumer;
   import org.apache.kafka.clients.producer.ProducerConfig;
   import org.apache.kafka.common.errors.WakeupException;

   public class KafkaMultiConsumerDemo {
       public static void main(String args[]) throws InterruptedException {
           //加载kafka.properties。
           Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

           Properties props = new Properties();
           //设置接入点，请通过控制台获取对应Topic的接入点。
           props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
           //两次Poll之间的最大允许间隔。
           //消费者超过该值没有返回心跳，服务端判断消费者处于非存活状态，服务端将消费者从Group移除并触发Rebalance，默认30s。
           props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
           //每次Poll的最大数量。
           //注意该值不要改得太大，如果Poll太多数据，而不能在下次Poll之前消费完，则会触发一次负载均衡，产生卡顿。
           props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
           //消息的反序列化方式。
           props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
           props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
           //当前消费实例所属的消费组，请在控制台申请之后填写。
           //属于同一个组的消费实例，会负载消费消息。
           props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));

           int consumerNum = 2;
           Thread[] consumerThreads = new Thread[consumerNum];
           for (int i = 0; i < consumerNum; i++) {
               KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

               List<String> subscribedTopics = new ArrayList<String>();
               subscribedTopics.add(kafkaProperties.getProperty("topic"));
               consumer.subscribe(subscribedTopics);

               KafkaConsumerRunner kafkaConsumerRunner = new KafkaConsumerRunner(consumer);
               consumerThreads[i] = new Thread(kafkaConsumerRunner);
           }

           for (int i = 0; i < consumerNum; i++) {
               consumerThreads[i].start();
           }

           for (int i = 0; i < consumerNum; i++) {
               consumerThreads[i].join();
           }
       }

       static class KafkaConsumerRunner implements Runnable {
           private final AtomicBoolean closed = new AtomicBoolean(false);
           private final KafkaConsumer consumer;

           KafkaConsumerRunner(KafkaConsumer consumer) {
               this.consumer = consumer;
           }

           @Override
           public void run() {
               try {
                   while (!closed.get()) {
                       try {
                           ConsumerRecords<String, String> records = consumer.poll(1000);
                           //必须在下次Poll之前消费完这些数据, 且总耗时不得超过SESSION_TIMEOUT_MS_CONFIG。
                           for (ConsumerRecord<String, String> record : records) {
                               System.out.println(String.format("Thread:%s Consume partition:%d offset:%d", Thread.currentThread().getName(), record.partition(), record.offset()));
                           }
                       } catch (Exception e) {
                           try {
                               Thread.sleep(1000);
                           } catch (Throwable ignore) {

                           }
                           e.printStackTrace();
                       }
                   }
               } catch (WakeupException e) {
                   //如果关闭则忽略异常。
                   if (!closed.get()) {
                       throw e;
                   }
               } finally {
                   consumer.close();
               }
           }
           //可以被另一个线程调用的关闭Hook。
           public void shutdown() {
               closed.set(true);
               consumer.wakeup();
           }
       }
   }