Kafka入门实例

谁借莪１个温暖的怀抱￠ 2021-09-11 01:52 364阅读 0赞

摘要:本文主要讲了Kafka的一个简单入门实例

源码下载：[https://github.com/appleappleapple/BigDataLearning][https_github.com_appleappleapple_BigDataLearning]

kafka安装过程看这里：[Kafka在Windows安装运行][Kafka_Windows]

整个工程目录如下：

![20160910125754528][]

1、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>
    	<groupId>com.lin</groupId>
    	<artifactId>Kafka-Demo</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.apache.kafka</groupId>
    			<artifactId>kafka_2.10</artifactId>
    			<version>0.9.0.0</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.opentsdb</groupId>
    			<artifactId>java-client</artifactId>
    			<version>2.1.0-SNAPSHOT</version>
    			<exclusions>
    				<exclusion>
    					<groupId>org.slf4j</groupId>
    					<artifactId>slf4j-log4j12</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>log4j</groupId>
    					<artifactId>log4j</artifactId>
    				</exclusion>
    				<exclusion>
    					<groupId>org.slf4j</groupId>
    					<artifactId>jcl-over-slf4j</artifactId>
    				</exclusion>
    			</exclusions>
    		</dependency>
    
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    			<version>1.2.4</version>
    		</dependency>
    
    
    	</dependencies>
    </project>

2、生产者

package com.lin.demo.producer;
    
    import java.util.Properties;
    
    import kafka.javaapi.producer.Producer;
    import kafka.producer.KeyedMessage;
    import kafka.producer.ProducerConfig;
    
    public class KafkaProducer {
    	private final Producer<String, String> producer;
    	public final static String TOPIC = "linlin";
    
    	private KafkaProducer() {
    		Properties props = new Properties();
    		// 此处配置的是kafka的端口
    		props.put("metadata.broker.list", "127.0.0.1:9092");
    		props.put("zk.connect", "127.0.0.1:2181");  
    
    		// 配置value的序列化类
    		props.put("serializer.class", "kafka.serializer.StringEncoder");
    		// 配置key的序列化类
    		props.put("key.serializer.class", "kafka.serializer.StringEncoder");
    
    		props.put("request.required.acks", "-1");
    
    		producer = new Producer<String, String>(new ProducerConfig(props));
    	}
    
    	void produce() {
    		int messageNo = 1000;
    		final int COUNT = 10000;
    
    		while (messageNo < COUNT) {
    			String key = String.valueOf(messageNo);
    			String data = "hello kafka message " + key;
    			producer.send(new KeyedMessage<String, String>(TOPIC, key, data));
    			System.out.println(data);
    			messageNo++;
    		}
    	}
    
    	public static void main(String[] args) {
    		new KafkaProducer().produce();
    	}
    }

右键：run as java application

运行结果：

![20160910130110420][]

3、消费者

package com.lin.demo.consumer;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.Properties;
    
    import kafka.consumer.ConsumerConfig;
    import kafka.consumer.ConsumerIterator;
    import kafka.consumer.KafkaStream;
    import kafka.javaapi.consumer.ConsumerConnector;
    import kafka.serializer.StringDecoder;
    import kafka.utils.VerifiableProperties;
    
    import com.lin.demo.producer.KafkaProducer;
    
    public class KafkaConsumer {
    
    	private final ConsumerConnector consumer;
    
    	private KafkaConsumer() {
    		Properties props = new Properties();
    		// zookeeper 配置
    		props.put("zookeeper.connect", "127.0.0.1:2181");
    
    		// group 代表一个消费组
    		props.put("group.id", "lingroup");
    
    		// zk连接超时
    		props.put("zookeeper.session.timeout.ms", "4000");
    		props.put("zookeeper.sync.time.ms", "200");
    		props.put("rebalance.max.retries", "5");
    		props.put("rebalance.backoff.ms", "1200");
    		
    	
    		props.put("auto.commit.interval.ms", "1000");
    		props.put("auto.offset.reset", "smallest");
    		// 序列化类
    		props.put("serializer.class", "kafka.serializer.StringEncoder");
    
    		ConsumerConfig config = new ConsumerConfig(props);
    
    		consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
    	}
    
    	void consume() {
    		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    		topicCountMap.put(KafkaProducer.TOPIC, new Integer(1));
    
    		StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
    		StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());
    
    		Map<String, List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
    		KafkaStream<String, String> stream = consumerMap.get(KafkaProducer.TOPIC).get(0);
    		ConsumerIterator<String, String> it = stream.iterator();
    		while (it.hasNext())
    			System.out.println("<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<" + it.next().message() + "<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<");
    	}
    
    	public static void main(String[] args) {
    		new KafkaConsumer().consume();
    	}
    }

运行结果：

![20160910130305485][]

监控页面

![20160910130941923][]

源码下载：[https://github.com/appleappleapple/BigDataLearning][https_github.com_appleappleapple_BigDataLearning]

[https_github.com_appleappleapple_BigDataLearning]: https://github.com/appleappleapple/BigDataLearning
[Kafka_Windows]: http://blog.csdn.net/evankaka/article/details/52421314
[20160910125754528]: /images/20210811/bb4d52fac89c471abd7678d8e9437764.png
[20160910130110420]: /images/20210811/1f22d2b351bf423188825ff80054836b.png
[20160910130305485]: /images/20210811/f7774573bf4a4a8e81b4a78e8955a629.png
[20160910130941923]: /images/20210811/0de5975346d9400eaf10120fc75a9a50.png