设为首页 收藏本站
查看: 1007|回复: 0

[经验分享] kafka小试

[复制链接]
累计签到:2 天
连续签到:1 天
发表于 2017-5-23 15:00:55 | 显示全部楼层 |阅读模式
  我使用的kafka版本 kafka_2.8.0-0.8.1.1.tgz
  参考了官网手册http://kafka.apache.org/documentation.html#quickstart
  和http://blog.csdn.net/hxpjava1/article/details/19160665  版本低一下,里面有些代码不兼容


  • 下载kafka 地址http://mirrors.hust.edu.cn/apache/kafka/0.8.1.1/kafka_2.8.0-0.8.1.1.tgz

tar -xzf kafka_2.9.2-0.8.1.1.tgz
cd kafka_2.9.2-0.8.1.1
  2.启动服务
  首先要启动zookeeper      

bin/zookeeper-server-start.sh config/zookeeper.properties &
  启动kafaka

bin/kafka-server-start.sh config/server.properties &
  3.创建topic

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
  查看是否创建成功

bin/kafka-topics.sh --list --zookeeper localhost:2181
  4.发送消息

import java.util.*;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
public class TestProducer {
public static void main(String[] args) {
Properties props = new Properties();
props.put("metadata.broker.list", "test.kafka.com:9092");
props.put("serializer.class", "kafka.serializer.StringEncoder");
props.put("request.required.acks", "1");
ProducerConfig config = new ProducerConfig(props);
Producer<String, String> producer = new Producer<String, String>(config);
KeyedMessage<String, String> data = new KeyedMessage<String, String>("test", "key", "测试");
producer.send(data);
producer.close();
System.out.println("结束");
}
}
  5.接收消息

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
public class ConsumerSample {
public static void main(String[] args) {  
// specify some consumer properties  
Properties props = new Properties();  
props.put("group.id", "test-consumer-group");
props.put("zookeeper.connect", "test.kafka.com:2181");
props.put("zookeeper.session.timeout.ms", "400");
props.put("zookeeper.sync.time.ms", "200");
props.put("auto.commit.interval.ms", "1000");
// Create the connection to the cluster  
ConsumerConfig consumerConfig = new ConsumerConfig(props);  
ConsumerConnector consumerConnector = Consumer.createJavaConsumerConnector(consumerConfig);  
// create 4 partitions of the stream for topic “test-topic”, to allow 4 threads to consume  
HashMap<String, Integer> map = new HashMap<String, Integer>();  
map.put("test", 4);
Map<String, List<KafkaStream<byte[], byte[]>>> topicMessageStreams =  
consumerConnector.createMessageStreams(map);  
List<KafkaStream<byte[], byte[]>> streams = topicMessageStreams.get("test");  
// create list of 4 threads to consume from each of the partitions   
ExecutorService executor = Executors.newFixedThreadPool(4);  
// consume the messages in the threads  
for (final KafkaStream<byte[], byte[]> stream : streams) {  
executor.submit(new Runnable() {  
public void run() {  
for (MessageAndMetadata<byte[], byte[]> msgAndMetadata : stream) {
System.out.println("topic:"+msgAndMetadata.topic());
String tmp = new String(msgAndMetadata.message());
System.out.println("message key: " + new String(msgAndMetadata.key()));
System.out.println("message content: " + tmp);  
}  
}  
});  
}  
}
}
  6.注意的地方
  test.kafka.com  为域名映射,可以自己映射到自己的kafka的ip地址
  如果发送消息失败 看下防火墙是否关闭
  对于group.id可以查看config/consumer.properties的配置
  7.如果出现FailedToSendMessageException: Failed to send messages after 3 tries错误
       修改config/server.properties  链接zookeeper为
      zookeeper.connect=127.0.0.1:2181
      配置的时候最好通过域名映射添加topic
    8.maven配置文件

<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.0.0.Final</version>
</dependency>
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.14</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.9.2</artifactId>
<version>0.8.1.1</version>
</dependency>
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>2.9.3</version>
</dependency>
<dependency>
<groupId>com.yammer.metrics</groupId>
<artifactId>metrics-core</artifactId>
<version>2.2.0</version>
</dependency>
<dependency>
<groupId>com.101tec</groupId>
<artifactId>zkclient</artifactId>
<version>0.3</version>
</dependency>
</dependencies>
   

运维网声明 1、欢迎大家加入本站运维交流群:群②:261659950 群⑤:202807635 群⑦870801961 群⑧679858003
2、本站所有主题由该帖子作者发表,该帖子作者与运维网享有帖子相关版权
3、所有作品的著作权均归原作者享有,请您和我们一样尊重他人的著作权等合法权益。如果您对作品感到满意,请购买正版
4、禁止制作、复制、发布和传播具有反动、淫秽、色情、暴力、凶杀等内容的信息,一经发现立即删除。若您因此触犯法律,一切后果自负,我们对此不承担任何责任
5、所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其内容的准确性、可靠性、正当性、安全性、合法性等负责,亦不承担任何法律责任
6、所有作品仅供您个人学习、研究或欣赏,不得用于商业或者其他用途,否则,一切后果均由您自己承担,我们对此不承担任何法律责任
7、如涉及侵犯版权等问题,请您及时通知我们,我们将立即采取措施予以解决
8、联系人Email:admin@iyunv.com 网址:www.yunweiku.com

所有资源均系网友上传或者通过网络收集,我们仅提供一个展示、介绍、观摩学习的平台,我们不对其承担任何法律责任,如涉及侵犯版权等问题,请您及时通知我们,我们将立即处理,联系人Email:kefu@iyunv.com,QQ:1061981298 本贴地址:https://www.yunweiku.com/thread-379782-1-1.html 上篇帖子: Kafka(二) -- Kafka 用在哪 下篇帖子: kafka issues
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

扫码加入运维网微信交流群X

扫码加入运维网微信交流群

扫描二维码加入运维网微信交流群,最新一手资源尽在官方微信交流群!快快加入我们吧...

扫描微信二维码查看详情

客服E-mail:kefu@iyunv.com 客服QQ:1061981298


QQ群⑦:运维网交流群⑦ QQ群⑧:运维网交流群⑧ k8s群:运维网kubernetes交流群


提醒:禁止发布任何违反国家法律、法规的言论与图片等内容;本站内容均来自个人观点与网络等信息,非本站认同之观点.


本站大部分资源是网友从网上搜集分享而来,其版权均归原作者及其网站所有,我们尊重他人的合法权益,如有内容侵犯您的合法权益,请及时与我们联系进行核实删除!



合作伙伴: 青云cloud

快速回复 返回顶部 返回列表