写了个 Kafka 的 Java Demo 顺便记录下 ， 仅供参考
1. 创建 maven 项目
目录如下
2.POM 文件:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd "> <ModelVersion> 4.0.0 </modelversion> <puperid> kafka-maven </groupiDiD> <TarfactId> kafka-maven </stifactid> <siperd> 0.1-snapshot> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>0.10.1.1</version> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-common</artifactId> <version>2.2.0</version> </dependency> <dependency> <GrupidId> org.apache.hadoop </roupidId> <TRAFACTID> HADOOP-HDFS </ArtifactId> <Versão> 2.2.0 </sipers> </dependency> <pendence> <puperid> org.apache.hadoop </proupid> <TartifactId> Hadoop-client </Artifactid> <PuerpId> org.apache.hbase </frupid> <stifactId> hbase-client <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>2.2.0</version> </dependency> <dependency> <groupId>jdk.tools</groupId> <artifactId>jdk.tools</artifactId> <version>1.7</version> <scope>system</scope> <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath> </dependency> <dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> <version>4.3.6</version> </dependency> </dependencies> <build> <plugins> <plugin> <PuerpId> org.apache.maven.plugins </frupidId> <ArtifactId> maven-compiler-plugin </stutifactId> <frifiguration> <locer> 1.7 </ro-fonte> </Target> 1.7 </Target> </figuration> </flugin> </lubins> </filt> </Project> 3.Kafka 生产者 Kafkaproduce:
pacote com.lijie.produces; importar java.io.file; importar java.io.fileInputStream; importar java.util.properties; importar org.apache.kafka.clients.producer.callback; import org.kafka.clients.producer.kafkkkkback; importache.kafka.clients.producer.kafkkkkback; org.apache.kafka.clients.producer.producerrecord; importar org.apache.kafka.clients.producer.recordmetadata; importação org.slf4j.logger; importação org.slf4j.logtory; static {Properties = new Properties (); String Path = Kafkaproduces.class.getResource ("/"). GetFile (). Tostring () + "kafka.properties"; tente {fileInputStream fis = new FileInputStream (novo arquivo (caminho)); Propriedades.load (fis); } catch (Exceção e) {e.printStackTrace (); }} / ** * 发送消息 * * @param tópico * @param key * @param value * / public void sendmsg (tópico da string, byte [] key, byte [] value) {// 实例化 produz kafkaproducer <byte [], byte []> kp = new Kafkaproducer <] // 消息封装 ProductErRecord <byte [], byte []> pr = new ProdproderRecord <byte [], byte []> (tópico, chave, valor); // 发送数据 kp.send (pr, new callback () {// 回调函数 @Override public void Oncompletion (registroMetadata Metadata, exceção de exceção) {if (null! = Excepctata) {System.out.println ("记录的 Offset 在:" + metadata.offset ()); }); // 关闭 Produce kp.close (); }} 4.Kafka 消费者 Kafkaconsume:
pacote com.lijie.consumer; importar java.io.file; importar java.io.fileInputStream; importar java.util.hashmap; importar java.util.list; import java.util.map; import java.util.moperties; com.lijie.pojo.User;import com.lijie.utils.JsonUtils;import kafka.consumer.ConsumerConfig;import kafka.consumer.ConsumerIterator;import kafka.consumer.KafkaStream;import kafka.javaapi.consumer.ConsumerConnector;import kafka.serializer.stringDecoder; importar kafka.utils.VerifiableProperties; classe pública kafkaconsume {private final Static String tópico = "lijietest"; Propriedades estáticas privadas; static {Properties = new Properties (); String Path = kafkaconsume.class.getResource ("/"). Getfile (). Tostring () + "kafka.properties"; tente {fileInputStream fis = new FileInputStream (novo arquivo (caminho)); Propriedades.load (fis); } catch (Exceção e) {e.printStackTrace (); }} / ** * 获取消息 * * @Throws Exception * / public void getMsg () lança exceção {consumerConfig config = new ConsumConfig (Propriedades); ConsumerConnector Consumer = kafka.consumer.consumer .CreateJavacOnsumerConnector (config); Mapa <String, Integer> tópicoCountMap = new Hashmap <String, Integer> (); tópicoCountMap.put (tópico, novo número inteiro (1)); StringDecoder keyDecoder = new StringDecoder (new VerifiableProperties ()); StringDecoder valueDecoder = new StringDecoder (new VerifiableProperties ()); Mapa <string, list <kafkAstream <string, string >>> ConsumerMap = consumer .createmessageStreamams (tópicoCountMap, keydecoder, valueDecoder); KafkAstream <string, string> stream = consumermap.get (tópico) .get (0); ConsumerIterator <string, string> it = stream.iterator (); while (it.hasnext ()) {string json = it.Next (). message (); Usuário do usuário = (usuário) jsonutils.jsontoobj (json, user.class); System.out.println (usuário); }}} 5.kafka.properties 文件
## ProduceBootstrap.Servers = 192.168.80.123: 9092Producer.type = syncrequest.required.acks = 1Serializer.class = kafka.serializer.defaultEncO derkey.Serializer = org.apache.kafka.common.serialization.bytearrayserializervalue.serializer = org.apache.kafka.common.serialization.byt earRaySerializerBak.Partitioner.class = kafka.producer.defaultPartitionerbak.key.serializer = org.apache.kafka.common.serialization.stri ngSerializerBak.value.Serializer = org.apache.kafka.common.serialization.stringSerializer ## consumezookeeper.connect = 192.168.80.123: 2181 group.id = lijieGroup zookeeper.session.Timeout.MS = 4000 zookeeper.sync.time.ms = 200 auto.commit.interval.ms = 1000 auto.offset.reset = menor serializador.class = kafka.serializer.stringEncoder
以上就是本文的全部内容 ， 希望对大家的学习有所帮助 ， 也希望大家多多支持武林网。