kafka版本：kafka_2.11-0.11.0.1.tgz

centos7

關(guān)閉防火墻：systemctl stop firewalld.service

在kafka中配置文件：

listeners=PLAINTEXT://192.168.80.132:9092

在java創(chuàng)建consumer：

package com.vincent;import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer;import java.util.Arrays; import java.util.Properties;/*** Created with IntelliJ IDEA.* User: vincent* Date: 2017/10/26* Time: 14:39*/ public class Test {public static void main(String[] args) {Properties props = new Properties();props.put("bootstrap.servers", "192.168.80.132:9092");props.put("group.id", "test");props.put("enable.auto.commit", "true");props.put("auto.commit.interval.ms", "1000");props.put("session.timeout.ms", "30000");props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);consumer.subscribe(Arrays.asList("test"));while (true) {ConsumerRecords<String, String> records = consumer.poll(100);for (ConsumerRecord<String, String> record : records)System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());}} }

?

總結(jié)

以上是生活随笔為你收集整理的java客户端访问kafka的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。