前言
在上一篇中讲述如何搭建kafka集群，本篇则讲述如何简单的使用 kafka 。不过在使用kafka的时候，还是应该简单的了解下kafka。
kafka的介绍
kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。
kafka 有如下特性：
- 以时间复杂度为o(1)的方式提供消息持久化能力，即使对tb级以上数据也能保证常数时间复杂度的访问性能。
- 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100k条以上消息的传输。
- 支持kafka server间的消息分区，及分布式消费，同时保证每个partition内的消息顺序传输。
- 同时支持离线数据处理和实时数据处理。
- scale out：支持在线水平扩展。
kafka的术语
- broker：kafka集群包含一个或多个服务器，这种服务器被称为broker。
- topic：每条发布到kafka集群的消息都有一个类别，这个类别被称为topic。（物理上不同topic的消息分开存储，逻辑上一个topic的消息虽然保存于一个或多个broker上但用户只需指定消息的topic即可生产或消费数据而不必关心数据存于何处）
- partition：partition是物理上的概念，每个topic包含一个或多个partition。
- producer：负责发布消息到kafka broker。
- consumer：消息消费者，向kafka broker读取消息的客户端。
- consumer group:每个consumer属于一个特定的consumer group（可为每个consumer指定group name，若不指定group name则属于默认的group）。
kafka核心api
kafka有四个核心api
- 应用程序使用producer api发布消息到1个或多个topic中。
- 应用程序使用consumer api来订阅一个或多个topic,并处理产生的消息。
- 应用程序使用streams api充当一个流处理器,从1个或多个topic消费输入流,并产生一个输出流到1个或多个topic,有效地将输入流转换到输出流。
- connector api允许构建或运行可重复使用的生产者或消费者,将topic链接到现有的应用程序或数据系统。
示例图如下:
kafka 应用场景
- 构建可在系统或应用程序之间可靠获取数据的实时流数据管道。
- 构建实时流应用程序，可以转换或响应数据流。
以上介绍参考kafka官方文档。
开发准备
如果我们要开发一个kafka的程序，应该做些什么呢？
首先，在搭建好kafka环境之后，我们要考虑的是我们是生产者还是消费者，也就是消息的发送者还是接受者。
不过在本篇中，生产者和消费者都会进行开发和讲解。
在大致的了解kafka之后，我们来开发第一个程序。
这里用的开发语言是java，构建工具maven。
maven的依赖如下:
?
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
<dependency>
<groupid>org.apache.kafka</groupid>
<artifactid>kafka_2.12</artifactid>
<version>1.0.0</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupid>org.apache.kafka</groupid>
<artifactid>kafka-clients</artifactid>
<version>1.0.0</version>
</dependency>
<dependency>
<groupid>org.apache.kafka</groupid>
<artifactid>kafka-streams</artifactid>
<version>1.0.0</version>
</dependency>
|
kafka producer
在开发生产的时候，先简单的介绍下kafka各种配置说明：
- bootstrap.servers： kafka的地址。
- acks:消息的确认机制，默认值是0。
- acks=0：如果设置为0，生产者不会等待kafka的响应。
- acks=1：这个配置意味着kafka会把这条消息写到本地日志文件中，但是不会等待集群中其他机器的成功响应。
- acks=all：这个配置意味着leader会等待所有的follower同步完成。这个确保消息不会丢失，除非kafka集群中所有机器挂掉。这是最强的可用性保证。
- retries：配置为大于0的值的话，客户端会在消息发送失败时重新发送。
- batch.size:当多条消息需要发送到同一个分区时，生产者会尝试合并网络请求。这会提高client和生产者的效率。
- key.serializer: 键序列化，默认org.apache.kafka.common.serialization.stringdeserializer。
- value.deserializer:值序列化，默认org.apache.kafka.common.serialization.stringdeserializer。
…
还有更多配置，可以去查看官方文档，这里就不在说明了。
那么我们kafka 的producer配置如下:
?
|
1
2
3
4
5
6
7
8
|
properties props = new properties();
props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("key.serializer", stringserializer.class.getname());
props.put("value.serializer", stringserializer.class.getname());
kafkaproducer<string, string> producer = new kafkaproducer<string, string>(props);
|
kafka的配置添加之后，我们便开始生产数据，生产数据代码只需如下就行:
?
|
1
|
producer.send(new producerrecord<string, string>(topic,key,value));
|
- topic: 消息队列的名称，可以先行在kafka服务中进行创建。如果kafka中并未创建该topic，那么便会自动创建！
- key:键值，也就是value对应的值，和map类似。
- value:要发送的数据，数据格式为string类型的。
在写好生产者程序之后，那我们先来生产吧！
我这里发送的消息为:
?
|
1
|
string messagestr="你好，这是第"+messageno+"条数据";
|
并且只发送1000条就退出，结果如下:
可以看到信息成功的打印了。
如果不想用程序进行验证程序是否发送成功，以及消息发送的准确性，可以在kafka服务器上使用命令查看。
kafka consumer
kafka消费这块应该来说是重点，毕竟大部分的时候，我们主要使用的是将数据进行消费。
- bootstrap.servers： kafka的地址。
- group.id：组名 不同组名可以重复消费。例如你先使用了组名a消费了kafka的1000条数据，但是你还想再次进行消费这1000条数据，并且不想重新去产生，那么这里你只需要更改组名就可以重复消费了。
- enable.auto.commit：是否自动提交，默认为true。
- auto.commit.interval.ms: 从poll(拉)的回话处理时长。
- session.timeout.ms:超时时间。
- max.poll.records:一次最大拉取的条数。
-
auto.offset.reset：消费规则，默认earliest 。
earliest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 。
latest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 。
none: topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常。 - key.serializer: 键序列化，默认org.apache.kafka.common.serialization.stringdeserializer。
- value.deserializer:值序列化，默认org.apache.kafka.common.serialization.stringdeserializer。
那么我们kafka 的consumer配置如下:
?
|
1
2
3
4
5
6
7
8
9
10
11
|
properties props = new properties();
props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
props.put("group.id", groupid);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("max.poll.records", 1000);
props.put("auto.offset.reset", "earliest");
props.put("key.deserializer", stringdeserializer.class.getname());
props.put("value.deserializer", stringdeserializer.class.getname());
kafkaconsumer<string, string> consumer = new kafkaconsumer<string, string>(props);
|
由于我这是设置的自动提交，所以消费代码如下:
我们需要先订阅一个topic，也就是指定消费哪一个topic。
?
|
1
|
consumer.subscribe(arrays.aslist(topic));
|
订阅之后，我们再从kafka中拉取数据:
?
|
1
|
consumerrecords<string, string> msglist=consumer.poll(1000);
|
一般来说进行消费会使用监听，这里我们就用for(;;)来进行监听， 并且设置消费1000条就退出！
结果如下:
可以看到我们这里已经成功消费了生产的数据了。
代码
那么生产者和消费者的代码如下:
生产者:
?
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
|
import java.util.properties;
import org.apache.kafka.clients.producer.kafkaproducer;
import org.apache.kafka.clients.producer.producerrecord;
import org.apache.kafka.common.serialization.stringserializer;
/**
*
* title: kafkaproducertest
* description:
* kafka 生产者demo
* version:1.0.0
* @author pancm
* @date 2018年1月26日
*/
public class kafkaproducertest implements runnable {
private final kafkaproducer<string, string> producer;
private final string topic;
public kafkaproducertest(string topicname) {
properties props = new properties();
props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("key.serializer", stringserializer.class.getname());
props.put("value.serializer", stringserializer.class.getname());
this.producer = new kafkaproducer<string, string>(props);
this.topic = topicname;
}
@override
public void run() {
int messageno = 1;
try {
for(;;) {
string messagestr="你好，这是第"+messageno+"条数据";
producer.send(new producerrecord<string, string>(topic, "message", messagestr));
//生产了100条就打印
if(messageno%100==0){
system.out.println("发送的信息:" + messagestr);
}
//生产1000条就退出
if(messageno%1000==0){
system.out.println("成功发送了"+messageno+"条");
break;
}
messageno++;
}
} catch (exception e) {
e.printstacktrace();
} finally {
producer.close();
}
}
public static void main(string args[]) {
kafkaproducertest test = new kafkaproducertest("kafka_test");
thread thread = new thread(test);
thread.start();
}
}
|
消费者:
?
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
|
import java.util.arrays;
import java.util.properties;
import org.apache.kafka.clients.consumer.consumerrecord;
import org.apache.kafka.clients.consumer.consumerrecords;
import org.apache.kafka.clients.consumer.kafkaconsumer;
import org.apache.kafka.common.serialization.stringdeserializer;
/**
*
* title: kafkaconsumertest
* description:
* kafka消费者 demo
* version:1.0.0
* @author pancm
* @date 2018年1月26日
*/
public class kafkaconsumertest implements runnable {
private final kafkaconsumer<string, string> consumer;
private consumerrecords<string, string> msglist;
private final string topic;
private static final string groupid = "groupa";
public kafkaconsumertest(string topicname) {
properties props = new properties();
props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
props.put("group.id", groupid);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");
props.put("auto.offset.reset", "earliest");
props.put("key.deserializer", stringdeserializer.class.getname());
props.put("value.deserializer", stringdeserializer.class.getname());
this.consumer = new kafkaconsumer<string, string>(props);
this.topic = topicname;
this.consumer.subscribe(arrays.aslist(topic));
}
@override
public void run() {
int messageno = 1;
system.out.println("---------开始消费---------");
try {
for (;;) {
msglist = consumer.poll(1000);
if(null!=msglist&&msglist.count()>0){
for (consumerrecord<string, string> record : msglist) {
//消费100条就打印 ,但打印的数据不一定是这个规律的
if(messageno%100==0){
system.out.println(messageno+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
}
//当消费了1000条就退出
if(messageno%1000==0){
break;
}
messageno++;
}
}else{
thread.sleep(1000);
}
}
} catch (interruptedexception e) {
e.printstacktrace();
} finally {
consumer.close();
}
}
public static void main(string args[]) {
kafkaconsumertest test1 = new kafkaconsumertest("kafka_test");
thread thread1 = new thread(test1);
thread1.start();
}
}
|
注: master、slave1、slave2 是因为我在自己的环境做了关系映射，这个可以换成服务器的ip。
当然项目我放在github上了，有兴趣的可以看看。https://github.com/xuwujing/kafka
总结
简单的开发一个kafka的程序需要以下步骤:
kafka介绍参考官方文档:http://kafka.apache.org/intro
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对快网idc的支持。
原文链接：https://www.cnblogs.com/xuwujing/p/8371127.html
相关文章
猜你喜欢
- 64M VPS建站：能否支持高流量网站运行？ 2025-06-10
- 64M VPS建站：怎样选择合适的域名和SSL证书？ 2025-06-10
- 64M VPS建站：怎样优化以提高网站加载速度？ 2025-06-10
- 64M VPS建站：是否适合初学者操作和管理？ 2025-06-10
- ASP.NET自助建站系统中的用户注册和登录功能定制方法 2025-06-10
