效果器,ca1102,911电影网
前言
在中讲述如何搭建kafka集群,本篇则讲述如何简单的使用 kafka 。不过在使用kafka的时候,还是应该简单的了解下kafka。
kafka的介绍
kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。
kafka 有如下特性:
kafka的术语
kafka核心api
kafka有四个核心api
示例图如下:
kafka 应用场景
以上介绍参考kafka官方文档。
开发准备
如果我们要开发一个kafka的程序,应该做些什么呢?
首先,在搭建好kafka环境之后,我们要考虑的是我们是生产者还是消费者,也就是消息的发送者还是接受者。
不过在本篇中,生产者和消费者都会进行开发和讲解。
在大致的了解kafka之后,我们来开发第一个程序。
这里用的开发语言是java,构建工具maven。
maven的依赖如下:
<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各种配置说明:
...
还有更多配置,可以去查看官方文档,这里就不在说明了。
那么我们kafka 的producer配置如下:
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的配置添加之后,我们便开始生产数据,生产数据代码只需如下就行:
producer.send(new producerrecord<string, string>(topic,key,value));
在写好生产者程序之后,那我们先来生产吧!
我这里发送的消息为:
string messagestr="你好,这是第"+messageno+"条数据";
并且只发送1000条就退出,结果如下:
可以看到信息成功的打印了。
如果不想用程序进行验证程序是否发送成功,以及消息发送的准确性,可以在kafka服务器上使用命令查看。
kafka consumer
kafka消费这块应该来说是重点,毕竟大部分的时候,我们主要使用的是将数据进行消费。
kafka消费的配置如下:
那么我们kafka 的consumer配置如下:
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。
consumer.subscribe(arrays.aslist(topic));
订阅之后,我们再从kafka中拉取数据:
consumerrecords<string, string> msglist=consumer.poll(1000);
一般来说进行消费会使用监听,这里我们就用for(;;)来进行监听, 并且设置消费1000条就退出!
结果如下:
可以看到我们这里已经成功消费了生产的数据了。
代码
那么生产者和消费者的代码如下:
生产者:
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(); } }
消费者:
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上了,有兴趣的可以看看。 ()
总结
简单的开发一个kafka的程序需要以下步骤:
kafka介绍参考官方文档:
总结
以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,如果有疑问大家可以留言交流,谢谢大家对移动技术网的支持。
如对本文有疑问,请在下面进行留言讨论,广大热心网友会与你互动!! 点击进行留言回复
浅析我对 String、StringBuilder、StringBuffer 的理解
使用IDEA搭建SSM框架的详细教程(spring + springMVC +MyBatis)
Springboot整合freemarker 404问题解决方案
引入mybatis-plus报 Invalid bound statement错误问题的解决方法
网友评论