Kafaka入门篇
1.Apache Kafka - 简介
Apache Kafka是一个分布式发布 - 订阅消息系统和一个强大的队列,可以处理大量的数据,并使您能够将消息从一个端点传递到另一个端点。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上,并在群集内复制以防止数据丢失。 Kafka构建在ZooKeeper同步服务之上。 它与Apache Storm和Spark非常好地集成,用于实时流式数据分析。Kafka专为分布式高吞吐量系统而设计。 Kafka往往工作得很好,作为一个更传统的消息代理的替代品。 与其他消息传递系统相比,Kafka具有更好的吞吐量,内置分区,复制和固有的容错能力,这使得它非常适合大规模消息处理应用程序。Kafka的具体理论知识可以查看官方文档介绍,http://kafka.apache.org/。
2.Kafka的安装
首先确保已经安装过java环境,如果没有可以去官网下载对应版本的jdk。我用的是jdk-8u161-linux-x64.tar.gz,zookeeper-3.4.10.tar.gz,kafka_2.11-1.1.0.tgz。
然后需要安装zookeeper,
tar -zxf zookeeper-3.4.10.tar.gz
cd zookeeper-3.4.10
mkdir data
vi conf/zoo.cfg
tickTime=2000
dataDir=/opt/zookeeper-3.4.10/data
clientPort=2181
initLimit=5
syncLimit=2
启用zookeeper:bin/zkServer.sh start
最后就是Kafka的安装:
cd /opt
tar -zxf kafka_2.11-1.1.0.tgz
cd kafka_2.11-1.1.0
启动: bin/kafka-server-start.sh config/server.properties,你将看到如下信息
关闭的话输入 bin/kafka-server-stop.sh config/server.properties 即可。
后台启动命令,bin/kafka-server-start.sh config/server.properties 1>/dev/null 2>&1 &
3.代码实现生产者
新建一个maven项目(当然java项目或SpringBoot项目也可以),引入所需jar包,pom.xml如下。
<?xml version="1.0" encoding="UTF-8"?>
<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>cn.sp</groupId>
<artifactId>kafka-demo</artifactId>
<version>1.0-SNAPSHOT</version> <dependencies>
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.12</artifactId>
<version>1.1.0</version>
</dependency> <!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.25</version>
<!--<scope>test</scope>-->
</dependency> <!-- https://mvnrepository.com/artifact/log4j/log4j -->
<dependency>
<groupId>log4j</groupId>
<artifactId>log4j</artifactId>
<version>1.2.17</version>
</dependency> </dependencies> </project>
消息生产者代码:
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord; import java.util.Properties; /**
* Created by 2YSP on 2018/4/3.
*/
public class ProducerTest { public static void main(String[] args) {
if (args.length == 0) {
System.out.println("Enter topic name");
return;
} //Assign topicName to string variable
String topicName = args[0]; //create instance for properties to access producer configs
Properties props = new Properties(); //Assign localhost id 192.168.75.132
props.put("bootstrap.servers","192.168.75.132:9092"); props.put("acks","all"); //If the request fails, the producer can automatically retry,
props.put("retries",0); //Specify buffer size in config
props.put("batch.size",16384); 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<String, String>(props);
for (int i=0;i<10;i++){
producer.send(new ProducerRecord<String, String>(topicName,Integer.toString(i),Integer.toString(i))); }
System.out.println("message send successfully");
producer.close();
}
}
当然,这里也可以弄一个.properties文件,然后代码里加载配置,运行时需要添加主题参数,我用的是idea就比较简单了。
4.代码实现消费者
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer; import java.util.Arrays;
import java.util.Properties; /**
* Created by 2YSP on 2018/4/3.
*/
public class ConsumerGroup { public static void main(String[] args) {
if (args.length < 2){
System.out.println("Usage: consumer <topic> <groupname> ");
return;
} String topic = args[0];
String group = args[1]; Properties props = new Properties();
props.put("bootstrap.servers","192.168.75.132:9092");
props.put("group.id",group);
props.put("enable.auto.commit","true");
props.put("auto.commit.interval.ms","1000");
props.put("session.timeout.ms","30000");
//当各分区下有已提交的offset时,从提交的offset开始消费;无提交的offset时,从头开始消费
props.put("auto.offset.reset", "earliest");
props.put("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(props); consumer.subscribe(Arrays.asList(topic));
while (true){
ConsumerRecords<String, String> records = consumer.poll(200);
for(ConsumerRecord<String, String> record:records){
System.out.printf("============offset = %d,key = %s,value=%s\n",record.offset(),record.key(),record.value());
}
//提交已经拉取出来的offset,如果是手动模式下面,必须拉取之后提交,否则以后会拉取重复消息
consumer.commitSync(); try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
先运行生产者代码,再运行消费者代码,控制台输出如下:
5.遇到的坑
本以为很简单的,结果中间遇到了很多坑,所以想记录下。
坑一:kafka_2.11-0.9.0.0.tgz这个版本的kafka一运行发送消息命令(bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name)就卡死了,弄了半天也找不到原因,无奈换了版本。
坑二:引入日志jar包不全,运行报错,折腾了一会儿。
坑三:一运行代码就报连接不上的错误,抛异常。
解决办法:找到config文件夹下的server.properties文件,修改advertised.listeners=PLAINTEXT://192.168.75.132:9092,其中192.168.75.132是我虚拟机的ip地址,以便外网访问。
再找到consumer.properties文件,bootstrap.servers=192.168.75.132:9092,group.id=group1,这里的group.id也可以不改就用默认的,但是代码配置里的一定要一样。
坑四:然后还是无法连接,想到防火墙问题就准备改端口,发现centos7的/etc/sysconfig文件夹下没有iptables文件。。。
解决办法:照着网上的办法弄好了,连接地址,又觉得改配置麻烦就索性关闭防火墙了。
至此,程序终于跑通了。。。
参考文档:https://www.w3cschool.cn/apache_kafka/
最新文章
- android apk静默安装
- 【Android】AppCompat V21:将 Materia Design 兼容到5.0之前的设备
- IE6/IE7中li底部4px的Bug
- ural 1837 Isenbaev&#39;s Number
- 更新-----Scripts:执行双网卡绑定
- GitHub版本控制
- curl讲解第一篇---入门和基本使用
- go语言学习-常用命令(四)
- 自己绘制table,加分页
- (转)初学Git及简单搭建git服务器和客户端
- 龟速机器学习总结----day1
- vue2.0路由
- 最详尽的 JS 原型与原型链终极详解,没有「可能是」。(一)
- python的os模块中的os.walk()函数
- ajax实现给JavaScript中全局变量赋值(转)
- UnicodeString基本操作(Ring3)
- Artech的MVC4框架学习——第七章Action的执行
- 01_Spark基础
- poj3070 Fibonacci(矩阵快速幂)
- Why we should overwrite the hashCode() when we overwrite the equals()
热门文章
- com.sun.xxx.utils不存在问题的解决
- Java日志框架-Logback手册中文版以及官方配置文档教程
- MySQL 资源大全
- Office WORD里面打字,后面的字自动被删除怎么办
- c++中拷贝构造函数,浅拷贝和深拷贝的区别
- android真机调试 INSTALL_FAILED_MEDIA_UNAVAILABLE 问题解决方案
- 高清接口芯片---gv7600、sii9135
- hadoop打jar包
- C++的cout高阶格式化操作
- Koa2学习(六)使用koa-router