摘要:准备环境系统下载并解压启动启动启动创建我修改了默认的端口查看创建的查看的详细信息测试运行并在控制台中输一些消息测试注意需要开两个终端搭建开发环境生产者
准备环境
系统
CentOS release 6.6 (Final)
下载并解压
wget http://mirror.bit.edu.cn/apache/kafka/0.8.0/kafka_2.10-0.8.0.tgz tar -zxvf kafka_2.10-0.8.0.tgz`启动
启动Zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties &
启动kafka
bin/kafka-server-start.sh config/server.properties &
创建topic
我修改了默认的端口
bin/kafka-topics.sh --create --zookeeper localhost:2182 --replication-factor 1 --partitions 1 --topic testweixuan &
查看创建的topic
bin/kafka-topics.sh --list --zookeeper localhost:2182
查看topic的详细信息
bin/kafka-topics.sh --describe --zookeeper localhost:2182
测试producer
运行producer并在控制台中输一些消息:
bin/kafka-console-producer.sh --broker-list 192.168.1.116:2182:9092 --topic testweixuan
测试consumer
bin/kafka-console-consumer.sh --zookeeper 192.168.1.116:2182 --topic testweixuan --from-beginning
注意:需要开两个终端
搭建开发环境 生产者代码package com.fengtang; import java.util.Properties; import kafka.javaapi.producer.Producer; import kafka.producer.KeyedMessage; import kafka.producer.ProducerConfig; /** * Create by fengtang * 2015/10/8 0008 * KafkaDemo_01 */ public class KafkaProducer { private final Producer消费者代码producer; public final static String TOPIC = "TEST-TOPIC"; private KafkaProducer() { Properties props = new Properties(); /** * 此处配置的是kafka的端口 */ props.put("metadata.broker.list", "192.168.1.116:9092"); /** * 配置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<>(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<>(TOPIC, key, data)); System.out.println(data); messageNo++; } } public static void main(String[] args) { new KafkaProducer().produce(); } }
package com.fengtang; 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 java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Properties; /** * Create by fengtang * 2015/10/8 0008 * KafkaDemo_01 */ public class KafkaConsumer { private final ConsumerConnector consumer; private KafkaConsumer() { Properties props = new Properties(); /** * zookeeper 配置 */ props.put("zookeeper.connect", "192.168.1.116:2182"); /** * group 代表一个消费组 */ props.put("group.id", "jd-group"); /** * zk连接超时 */ props.put("zookeeper.session.timeout.ms", "400000"); props.put("zookeeper.sync.time.ms", "200"); 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() { MaptopicCountMap = new HashMap (); topicCountMap.put(KafkaProducer.TOPIC, new Integer(1)); StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties()); StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties()); Map >> consumerMap = consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder); KafkaStream stream = consumerMap.get(KafkaProducer.TOPIC).get(0); ConsumerIterator it = stream.iterator(); while (it.hasNext()) System.out.println(it.next().message()); } public static void main(String[] args) { new KafkaConsumer().consume(); } }
pom.xml
解决报错4.0.0 com.fengtang kafkademo 1.0-SNAPSHOT jar kafkademo http://maven.apache.org UTF-8 org.apache.kafka kafka_2.10 0.8.0 log4j log4j 1.2.15 jmxtools com.sun.jdmk jmxri com.sun.jmx jms javax.jms javax.mail
kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
需要改动config文件夹下的server.properties中的以下两个属性
zookeeper.connect=localhost:2181改成zookeeper.connect=192.168.1.116 (自己的服务器IP地址):2181
以及默认注释掉的 #host.name=localhost 改成 host.name=192.168.1.116 (自己的服务器IP地址)
成功截图文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。
转载请注明本文地址:https://www.ucloud.cn/yun/64659.html
摘要:项目地址前言大数据技术栈思维导图大数据常用软件安装指南一分布式文件存储系统分布式计算框架集群资源管理器单机伪集群环境搭建集群环境搭建常用命令的使用基于搭建高可用集群二简介及核心概念环境下的安装部署和命令行的基本使用常用操作分区表和分桶表视图 项目GitHub地址:https://github.com/heibaiying... 前 言 大数据技术栈思维导图 大数据常用软件安装指...
阅读 3982·2021-11-22 13:53
阅读 3584·2021-11-19 11:29
阅读 1166·2021-09-08 09:35
阅读 3125·2020-12-03 17:26
阅读 480·2019-08-29 16:06
阅读 2079·2019-08-26 13:50
阅读 1148·2019-08-23 18:32
阅读 2111·2019-08-23 18:12