kafka学习(1)linux下的安装和启动,以及Java示例代码
2017-08-24 20:00
726 查看
1. 安装
1.1 下载kafka并解压
wget http://mirror.bit.edu.cn/apache/kafka/0.11.0.0/kafka_2.11-0.11.0.0.tgz tar -zxvf kafka_2.11-0.11.0.0.tgz
1.2 修改配置
(1)配置zookeeper地址vim /data/kafka_2.11-0.11.0.0/config/server.properties
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183
(2)配置内存
vim /data/kafka_2.11-0.11.0.0/bin/kafka-server-start.sh
if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then export KAFKA_HEAP_OPTS="-Xmx256m -Xms256m" fi
(3)敲黑板!修改对外的IP地址,否则kafka在别的机器上无法发布和订阅
[root@iZ2zei0nwllapkwklisoncZ kafka_2.11-0.11.0.0]# vim config/server.properties
advertised.listeners=PLAINTEXT://123.56.13.70:9092
1.3 启动
sh /data/kafka_2.11-0.11.0.0/bin/kafka-server-start.sh /data/kafka_2.11-0.11.0.0/config/server.properties
2. 发送和订阅消息
2.1 创建主题
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
2.2 查看主题
[root@iZ2zei0nwllapkwklisoncZ kafka_2.11-0.11.0.0]# bin/kafka-topics.sh --list --zookeeper localhost:2181 test
2.3 发送消息produce
[root@iZ2zei0nwllapkwklisoncZ kafka_2.11-0.11.0.0]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test >我是高天悦 >
2.4 接收消息consume
[root@iZ2zei0nwllapkwklisoncZ kafka_2.11-0.11.0.0]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning this is a msg this is another msg 你好
3. Java API 代码
3.1 生产者
public static String testProduce() { Properties props = new Properties(); props.put("bootstrap.servers", KafkaConstants.BROKER_LIST); props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", 33554432); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); Producer<String, String> producer = new KafkaProducer<>(props); for (int i = 0; i < 100; i++) producer.send( new ProducerRecord<String, String>("test", Integer.toString(i), Integer.toString(i))); producer.close(); return "ok"; }
3.2 消费者
public static Object testConsume() { Properties props = new Properties(); props.put("bootstrap.servers", "123.56.13.70:9092"); props.put("group.id", "d2"); props.put("enable.auto.commit", "false"); props.put("auto.commit.interval.ms", "1000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("session.timeout.ms", "30000"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); // consumer.subscribe(Arrays.asList("foo", "bar", "test")); consumer.subscribe(Arrays.asList("test")); long timeout = 10000; while (true) { ConsumerRecords<String, String> records = consumer.poll(timeout); for (ConsumerRecord<String, String> record : records) System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } }
在控制台和Eclipse console同时看到生产和消费之间的通讯,bingo!
tips: kafka的持久化在/tmp/kafka-logs下
[root@iZ2zei0nwllapkwklisoncZ kafka-logs]# pwd
/tmp/kafka-logs
[root@iZ2zei0nwllapkwklisoncZ kafka-logs]# ll
total 232
drwxr-xr-x 2 root root 4096 Oct 15 19:07 baotest-0
drwxr-xr-x 2 root root 4096 Oct 15 19:07 bar-0
-rw-r–r– 1 root root 0 Oct 14 21:32 cleaner-offset-checkpoint
drwxr-xr-x 2 root root 4096 Oct 15 19:23 __consumer_offsets-0
drwxr-xr-x 2 root root 4096 Oct 15 19:07 __consumer_offsets-1
….
drwxr-xr-x 2 root root 4096 Oct 15 19:07 __consumer_offsets-5
drwxr-xr-x 2 root root 4096 Oct 15 19:07 __consumer_offsets-6
drwxr-xr-x 2 root root 4096 Oct 15 19:07 __consumer_offsets-7
drwxr-xr-x 2 root root 4096 Oct 15 19:07 __consumer_offsets-8
drwxr-xr-x 2 root root 4096 Oct 15 19:07 __consumer_offsets-9
drwxr-xr-x 2 root root 4096 Oct 15 19:07 foo-0
-rw-r–r– 1 root root 4 Oct 15 20:47 log-start-offset-checkpoint
-rw-r–r– 1 root root 54 Oct 14 21:33 meta.properties
-rw-r–r– 1 root root 1244 Oct 15 20:47 recovery-point-offset-checkpoint
-rw-r–r– 1 root root 1248 Oct 15 20:48 replication-offset-checkpoint
drwxr-xr-x 2 root root 4096 Oct 15 19:15 test-0
[root@iZ2zei0nwllapkwklisoncZ kafka-logs]#
相关文章推荐
- linux(ubuntu)下安装和启动kafka以及若干坑
- JAVA学习笔记——第二章 Eclipse的下载安装以及配置和启动
- mongodb 的安装以及在多台服务器之间建立副本集 在虚拟机上安装多个linux实现 配合Java代码实现读写分离
- kafka介绍,安装以及简单的java调用kafka代码
- 安装部署(六) Kafka集群安装部署以及Producer和Consumer的JAVA代码测试
- IMAP协议指令学习以及代码(C/C++)示例
- ARM&LINUX学习笔记(6)------启动代码分析
- linux下memcached安装以及启动
- [学习指导] linux 启动过程以及 /etc/rc.d/init.d/目录的一点理解
- Linux下MemCached安装和启动以及C#调用
- Linux分区和挂载以及启动(mount命令的学习)
- linux下memcached安装以及启动
- LINUX安装tomcat 启动报异常 Protocol handler pause failed java.net.NoRouteToHostException: No route to host
- Java学习——Ubuntu下jdk的安装以及Java环境的配置
- 深入XPath的详解以及Java示例代码分析
- Java 学习笔记22:Maven以及其Eclipse插件m2eclipse的安装
- linux下memcached安装以及启动
- Linux操作系统: JDK 安装以及Java环境配置方法
- Java基础之Map学习代码示例一:
- 在Linux下安装Eclipse以及C++和JAVA插件