【郑州校区】kafka笔记之操作集群的两种方式 - 黑马程序员技术交流社区

【郑州校区】kafka笔记之操作集群的两种方式

需求：订单系统，需要发送消息。后面后3个程序需要接受这个消息，并做后续的处理。

1）创建一个订单的topic。

bin/kafka-topics.sh --create --zookeeper zk01:2181 --replication-factor 1 --partitions 1 --topic order

2）编写代码启动一个生产者，生产数据

bin/kafka-console-producer.sh --broker-list kafka01:9092 --topic order

3）编写代码启动给一个消费者，消费数据

bin/kafka-console-consumer.sh --zookeeper zk01:2181 --from-beginning --topic order

1）java工程-maven，依赖。

<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.11.0.1</version>
</dependency>

1）编写代码-写生产者的代码

/**
* 订单的生产者代码
*/
public class OrderProducer {
public static void main(String[] args) throws InterruptedException {
      /* 1、连接集群，通过配置文件的方式
      * 2、发送数据-topic:order，value
      */
      Properties props = new Properties();
      props.put("bootstrap.servers", "node01:9092");
      props.put("acks", "all");
      props.put("retries", 0);
      props.put("batch.size", 16384);
      props.put("linger.ms", 1);
      props.put("buffer.memory", 33554432);
      props.put("key.serializer",
            "org.apache.kafka.common.serialization.StringSerializer");
      props.put("value.serializer",
            "org.apache.kafka.common.serialization.StringSerializer");
      KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(props);
      for (int i = 0; i < 1000; i++) {
         // 发送数据 ,需要一个producerRecord对象,最少参数 String topic, V value
         kafkaProducer.send(new ProducerRecord<String, String>("order", "订单信息！"+i));
         Thread.sleep(100);
      }
}
}

2）编写代码-写消费者的代码

/**
* 消费订单数据--- javaben.tojson
*/
public class OrderConsumer {
public static void main(String[] args) {
      // 1\连接集群
      Properties props = new Properties();
      props.put("bootstrap.servers", "node01:9092");
      props.put("group.id", "test");
      props.put("enable.auto.commit", "true");
      props.put("auto.commit.interval.ms", "1000");
      props.put("key.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
      props.put("value.deserializer",
            "org.apache.kafka.common.serialization.StringDeserializer");
      KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(props);
//       2、发送数据发送数据需要，订阅下要消费的topic。  order
      kafkaConsumer.subscribe(Arrays.asList("order"));
      while (true) {
         ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(100);// jdk queue offer插入、poll获取元素。 blockingqueue put插入原生，take获取元素
         for (ConsumerRecord<String, String> record : consumerRecords) {
            System.out.println("消费的数据为：" + record.value());
         }
      }
}
}

传智播客·黑马程序员郑州校区地址

河南省郑州市高新区长椿路11号大学科技园（西区）东门8号楼三层

联系电话 0371-56061160/61/62

来校路线地铁一号线梧桐街站A口出