kafka介紹和使用

1.1.?????? 主要功能

根據(jù)官網的介紹株憾，ApacheKafka?是一個分布式流媒體平臺，它主要有3種功能：

　　1：It lets you publish and subscribe to streams of records.發(fā)布和訂閱消息流捏卓，這個功能類似于消息隊列，這也是kafka歸類為消息隊列框架的原因

　　2：It lets you store streams of records in a fault-tolerant way.以容錯的方式記錄消息流愿待，kafka以文件的方式來存儲消息流

　　3：It lets you process streams of records as they occur.可以再消息發(fā)布的時候進行處理

?1.2.?????? 使用場景

1：Building real-time streaming data pipelines that reliably get data between systems or applications.在系統(tǒng)或應用程序之間構建可靠的用于傳輸實時數(shù)據(jù)的管道实撒，消息隊列功能

2：Building real-time streaming applications that transform or react to the streams of data疫剃。構建實時的流數(shù)據(jù)處理程序來變換或處理數(shù)據(jù)流钉疫，數(shù)據(jù)處理功能

1.3.?????? 詳細介紹

Kafka目前主要作為一個分布式的發(fā)布訂閱式的消息系統(tǒng)使用，下面簡單介紹一下kafka的基本機制

　　1.3.1 消息傳輸流程

　　　　Producer即生產者巢价，向Kafka集群發(fā)送消息牲阁，在發(fā)送消息之前，會對消息進行分類壤躲，即Topic城菊，上圖展示了兩個producer發(fā)送了分類為topic1的消息，另外一個發(fā)送了topic2的消息碉克。

　　　　Topic即主題凌唬，通過對消息指定主題可以將消息分類，消費者可以只關注自己需要的Topic中的消息

　　　　Consumer即消費者漏麦，消費者通過與kafka集群建立長連接的方式客税，不斷地從集群中拉取消息况褪，然后可以對這些消息進行處理。

　　　　從上圖中就可以看出同一個Topic下的消費者和生產者的數(shù)量并不是對應的更耻。

　　1.3.2 kafka服務器消息存儲策略

　　　　談到kafka的存儲测垛，就不得不提到分區(qū)，即partitions秧均，創(chuàng)建一個topic時食侮，同時可以指定分區(qū)數(shù)目，分區(qū)數(shù)越多目胡，其吞吐量也越大锯七，但是需要的資源也越多，同時也會導致更高的不可用性誉己，kafka在接收到生產者發(fā)送的消息之后起胰，會根據(jù)均衡策略將消息存儲到不同的分區(qū)中。

　　在每個分區(qū)中巫延，消息以順序存儲效五，最晚接收的的消息會最后被消費。

　　1.3.3 與生產者的交互

　　　　生產者在向kafka集群發(fā)送消息的時候炉峰，可以通過指定分區(qū)來發(fā)送到指定的分區(qū)中

　　　　也可以通過指定均衡策略來將消息發(fā)送到不同的分區(qū)中

　　　　如果不指定畏妖，就會采用默認的隨機均衡策略，將消息隨機的存儲到不同的分區(qū)中

　　1.3.4? 與消費者的交互

　　　　在消費者消費消息時疼阔，kafka使用offset來記錄當前消費的位置

　　　　在kafka的設計中戒劫，可以有多個不同的group來同時消費同一個topic下的消息，如圖婆廊，我們有兩個不同的group同時消費迅细，他們的的消費的記錄位置offset各不項目，不互相干擾淘邻。

　　　　對于一個group而言茵典，消費者的數(shù)量不應該多余分區(qū)的數(shù)量，因為在一個group中宾舅，每個分區(qū)至多只能綁定到一個消費者上统阿，即一個消費者可以消費多個分區(qū)，一個分區(qū)只能給一個消費者消費

　　　　因此筹我，若一個group中的消費者數(shù)量大于分區(qū)數(shù)量的話扶平，多余的消費者將不會收到任何消息。

2.?????? Kafka安裝與使用

2.1.?????? 下載

　　你可以在kafka官網?http://kafka.apache.org/downloads下載到最新的kafka安裝包蔬蕊，選擇下載二進制版本的tgz文件结澄，根據(jù)網絡狀態(tài)可能需要fq，這里我們選擇的版本是0.11.0.1，目前的最新版

2.2.?????? 安裝

　　Kafka是使用scala編寫的運行與jvm虛擬機上的程序麻献，雖然也可以在windows上使用呼巷，但是kafka基本上是運行在linux服務器上，因此我們這里也使用linux來開始今天的實戰(zhàn)赎瑰。

　　首先確保你的機器上安裝了jdk王悍，kafka需要java運行環(huán)境，以前的kafka還需要zookeeper餐曼，新版的kafka已經內置了一個zookeeper環(huán)境压储，所以我們可以直接使用

　　說是安裝，如果只需要進行最簡單的嘗試的話我們只需要解壓到任意目錄即可源譬，這里我們將kafka壓縮包解壓到/home目錄

2.3.?????? 配置

　　在kafka解壓目錄下下有一個config的文件夾集惋，里面放置的是我們的配置文件

　　consumer.properites 消費者配置，這個配置文件用于配置于2.5節(jié)中開啟的消費者踩娘，此處我們使用默認的即可

　　producer.properties 生產者配置刮刑，這個配置文件用于配置于2.5節(jié)中開啟的生產者，此處我們使用默認的即可

　　server.properties kafka服務器的配置养渴，此配置文件用來配置kafka服務器雷绢，目前僅介紹幾個最基礎的配置

broker.id 申明當前kafka服務器在集群中的唯一ID，需配置為integer,并且集群中的每一個kafka服務器的id都應是唯一的理卑，我們這里采用默認配置即可

listeners 申明此kafka服務器需要監(jiān)聽的端口號翘紊，如果是在本機上跑虛擬機運行可以不用配置本項，默認會使用localhost的地址藐唠，如果是在遠程服務器上運行則必須配置帆疟，例如：

　　　　　　　　　　listeners=PLAINTEXT:// 192.168.180.128:9092。并確保服務器的9092端口能夠訪問

　　　　? 3.zookeeper.connect 申明kafka所連接的zookeeper的地址宇立，需配置為zookeeper的地址踪宠，由于本次使用的是kafka高版本中自帶? ? ? ? ? ? ? ??? ? ? ? zookeeper，使用默認配置即可?

　　　　　　　　　　zookeeper.connect=localhost:2181

2.4.?????? 運行

啟動zookeeper

cd進入kafka解壓目錄妈嘹，輸入

bin/zookeeper-server-start.sh config/zookeeper.properties

啟動zookeeper成功后會看到如下的輸出

　　　　2.啟動kafka

cd進入kafka解壓目錄柳琢，輸入

bin/kafka-server-start.sh config/server.properties

啟動kafka成功后會看到如下的輸出

2.5.?????? 第一個消息

　　　2.5.1? ?創(chuàng)建一個topic

　　　　Kafka通過topic對同一類的數(shù)據(jù)進行管理，同一類的數(shù)據(jù)使用同一個topic可以在處理數(shù)據(jù)時更加的便捷

　　　　在kafka解壓目錄打開終端蟋滴，輸入

　　　　bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic?test

　　　　創(chuàng)建一個名為test的topic

? ? ? ? 　在創(chuàng)建topic后可以通過輸入

? ? ? ? ? ??bin/kafka-topics.sh --list --zookeeper localhost:2181

???來查看已經創(chuàng)建的topic

　　2.4.2???創(chuàng)建一個消息消費者

　　　在kafka解壓目錄打開終端染厅，輸入

　　　　bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic?test?--from-beginning

　　　可以創(chuàng)建一個用于消費topic為test的消費者

???????? 消費者創(chuàng)建完成之后痘绎，因為還沒有發(fā)送任何數(shù)據(jù)津函，因此這里在執(zhí)行后沒有打印出任何數(shù)據(jù)

???????? 不過別著急，不要關閉這個終端孤页，打開一個新的終端尔苦，接下來我們創(chuàng)建第一個消息生產者

　　2.4.3? ? ? ? ?創(chuàng)建一個消息生產者

　　　　在kafka解壓目錄打開一個新的終端，輸入

　　　　bin/kafka-console-producer.sh --broker-list localhost:9092 --topic?test

　　　　在執(zhí)行完畢后會進入的編輯器頁面

在發(fā)送完消息之后，可以回到我們的消息消費者終端中允坚，可以看到魂那，終端中已經打印出了我們剛才發(fā)送的消息

3.?????? 使用java程序

　　　　跟上節(jié)中一樣，我們現(xiàn)在在java程序中嘗試使用kafka

　　　　3.1? 創(chuàng)建Topic

public static void main(String[] args) {

??? //創(chuàng)建topic

??? Properties props = new Properties();

??? props.put("bootstrap.servers", "192.168.180.128:9092");

??? AdminClient adminClient = AdminClient.create(props);

??? ArrayList<NewTopic> topics = new ArrayList<NewTopic>();

??? NewTopic newTopic = new NewTopic("topic-test", 1, (short) 1);

??? topics.add(newTopic);

??? CreateTopicsResult result = adminClient.createTopics(topics);

??? try {

??????? result.all().get();

??? } catch (InterruptedException e) {

??????? e.printStackTrace();

??? } catch (ExecutionException e) {

??????? e.printStackTrace();

??? }

}

　　使用AdminClient API可以來控制對kafka服務器進行配置稠项，我們這里使用NewTopic(String name, int numPartitions, short 　　replicationFactor)的構造方法來創(chuàng)建了一個名為“topic-test”涯雅，分區(qū)數(shù)為1，復制因子為1的Topic.

3.2 ?Producer生產者發(fā)送消息

public static void main(String[] args){

??? Properties props = new Properties();

??? props.put("bootstrap.servers", "192.168.180.128:9092");

??? 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<String, String>(props);

??? for (int i = 0; i < 100; i++)

??????? producer.send(new ProducerRecord<String, String>("topic-test", Integer.toString(i), Integer.toString(i)));

??? producer.close();

}

使用producer發(fā)送完消息可以通過2.5中提到的服務器端消費者監(jiān)聽到消息展运。也可以使用接下來介紹的java消費者程序來消費消息

3.3 Consumer消費者消費消息

public static void main(String[] args){

??? Properties props = new Properties();

??? props.put("bootstrap.servers", "192.168.12.65:9092");

??? props.put("group.id", "test");

??? props.put("enable.auto.commit", "true");

??? 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");

??? final KafkaConsumer<String, String> consumer = new KafkaConsumer<String,String>(props);

??? consumer.subscribe(Arrays.asList("topic-test"),new ConsumerRebalanceListener() {

??????? public void onPartitionsRevoked(Collection<TopicPartition> collection) {

??????? }

??????? public void onPartitionsAssigned(Collection<TopicPartition> collection) {

??????????? //將偏移設置到最開始

??????????? consumer.seekToBeginning(collection);

??????? }

??? });

??? while (true) {

??????? ConsumerRecords<String, String> records = consumer.poll(100);

??????? for (ConsumerRecord<String, String> record : records)

??????????? System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());

??? }

}

這里我們使用Consume API 來創(chuàng)建了一個普通的java消費者程序來監(jiān)聽名為“topic-test”的Topic活逆，每當有生產者向kafka服務器發(fā)送消息，我們的消費者就能收到發(fā)送的消息拗胜。

4.?????? 使用spring-kafka

Spring-kafka是正處于孵化階段的一個spring子項目蔗候，能夠使用spring的特性來讓我們更方便的使用kafka

4.1?? 基本配置信息

與其他spring的項目一樣，總是離不開配置埂软，這里我們使用java配置來配置我們的kafka消費者和生產者锈遥。

引入pom文件

??? <groupId>org.apache.kafka</groupId>

??? <artifactId>kafka-clients</artifactId>

??? <version>0.11.0.1</version>

</dependency>

??? <groupId>org.apache.kafka</groupId>

??? <artifactId>kafka-streams</artifactId>

??? <version>0.11.0.1</version>

</dependency>

??? <groupId>org.springframework.kafka</groupId>

??? <artifactId>spring-kafka</artifactId>

?? ?<version>1.3.0.RELEASE</version>

</dependency>

創(chuàng)建配置類

我們在主目錄下新建名為KafkaConfig的類

@Configuration

@EnableKafka

public class KafkaConfig {

}

配置Topic

在kafkaConfig類中添加配置

//topic config Topic的配置開始

??? @Bean

??? public KafkaAdmin admin() {

??????? Map<String, Object> configs = new HashMap<String, Object>();

??????? configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.180.128:9092");

??????? return new KafkaAdmin(configs);

??? }

??? @Bean

??? public NewTopic topic1() {

??????? return new NewTopic("foo", 10, (short) 2);

??? }

//topic的配置結束

配置生產者Factort及Template

//producer config start

??? @Bean

??? public ProducerFactory<Integer, String> producerFactory() {

??????? return new DefaultKafkaProducerFactory<Integer,String>(producerConfigs());

??? }

??? @Bean

??? public Map<String, Object> producerConfigs() {

??????? Map<String, Object> props = new HashMap<String,Object>();

??????? props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.180.128:9092");

??????? 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.IntegerSerializer");

??????? props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

??????? return props;

??? }

??? @Bean

??? public KafkaTemplate<Integer, String> kafkaTemplate() {

??????? return new KafkaTemplate<Integer, String>(producerFactory());

??? }

//producer config end

5.配置ConsumerFactory

//consumer config start

??? @Bean

??? public ConcurrentKafkaListenerContainerFactory<Integer,String> kafkaListenerContainerFactory(){

??????? ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<Integer, String>();

??????? factory.setConsumerFactory(consumerFactory());

??????? return factory;

??? }

??? @Bean

??? public ConsumerFactory<Integer,String> consumerFactory(){

??????? return new DefaultKafkaConsumerFactory<Integer, String>(consumerConfigs());

??? }

??? @Bean

??? public Map<String,Object> consumerConfigs(){

??????? HashMap<String, Object> props = new HashMap<String, Object>();

??????? props.put("bootstrap.servers", "192.168.180.128:9092");

??????? props.put("group.id", "test");

??????? props.put("enable.auto.commit", "true");

??????? props.put("auto.commit.interval.ms", "1000");

??????? props.put("key.deserializer", "org.apache.kafka.common.serialization.IntegerDeserializer");

??????? props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

??????? return props;

??? }

//consumer config end

4.2? 創(chuàng)建消息生產者

//使用spring-kafka的template發(fā)送一條消息發(fā)送多條消息只需要循環(huán)多次即可

public static void main(String[] args) throws ExecutionException, InterruptedException {

??? AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext(KafkaConfig.class);

??? KafkaTemplate<Integer, String> kafkaTemplate = (KafkaTemplate<Integer, String>) ctx.getBean("kafkaTemplate");

??????? String data="this is a test message";

??????? ListenableFuture<SendResult<Integer, String>> send = kafkaTemplate.send("topic-test", 1, data);

??????? send.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {

??????????? public void onFailure(Throwable throwable) {

??????????? }

??????????? public void onSuccess(SendResult<Integer, String> integerStringSendResult) {

??????????? }

??????? });

}

4.3??? 創(chuàng)建消息消費者

我們首先創(chuàng)建一個一個用于消息監(jiān)聽的類，當名為”topic-test”的topic接收到消息之后勘畔，我們的這個listen方法就會調用所灸。

public class SimpleConsumerListener {

??? private final static Logger?logger?= LoggerFactory.getLogger(SimpleConsumerListener.class);

??? private final CountDownLatch latch1 = new CountDownLatch(1);

??? @KafkaListener(id = "foo", topics = "topic-test")

??? public void listen(byte[] records) {

??????? //do something here

??????? this.latch1.countDown();

??? }

}

???????? 我們同時也需要將這個類作為一個Bean配置到KafkaConfig中

@Bean

public SimpleConsumerListener simpleConsumerListener(){

??? return new SimpleConsumerListener();

}

默認spring-kafka會為每一個監(jiān)聽方法創(chuàng)建一個線程來向kafka服務器拉取消息

————————————————

原文鏈接：https://blog.csdn.net/u012129558/article/details/80065869

最后編輯于：2019.10.13 20:33:39

?著作權歸作者所有,轉載或內容合作請聯(lián)系作者

人面猴
序言：七十年代末，一起剝皮案震驚了整個濱河市诉字，隨后出現(xiàn)的幾起案子懦尝，更是在濱河造成了極大的恐慌，老刑警劉巖壤圃，帶你破解...
沈念sama閱讀 217,406評論 6贊 503
死咒
序言：濱河連續(xù)發(fā)生了三起死亡事件陵霉，死亡現(xiàn)場離奇詭異，居然都是意外死亡伍绳，警方通過查閱死者的電腦和手機踊挠，發(fā)現(xiàn)死者居然都...
沈念sama閱讀 92,732評論 3贊 393
救了他兩次的神仙讓他今天三更去死
文/潘曉璐我一進店門，熙熙樓的掌柜王于貴愁眉苦臉地迎上來冲杀，“玉大人效床，你說我怎么就攤上這事∪ㄋ” “怎么了剩檀？”我有些...
開封第一講書人閱讀 163,711評論 0贊 353
道士緝兇錄：失蹤的賣姜人
文/不壞的土叔我叫張陵，是天一觀的道長旺芽。經常有香客問我沪猴，道長辐啄，這世上最難降的妖魔是什么？我笑而不...
開封第一講書人閱讀 58,380評論 1贊 293
?港島之戀（遺憾婚禮）
正文為了忘掉前任运嗜，我火速辦了婚禮壶辜，結果婚禮上，老公的妹妹穿的比我還像新娘担租。我一直安慰自己砸民，他們只是感情好，可當我...
茶點故事閱讀 67,432評論 6贊 392
惡毒庶女頂嫁案：這布局不是一般人想出來的
文/花漫我一把揭開白布奋救。她就那樣靜靜地躺著阱洪，像睡著了一般。火紅的嫁衣襯著肌膚如雪菠镇。梳的紋絲不亂的頭發(fā)上冗荸，一...
開封第一講書人閱讀 51,301評論 1贊 301
城市分裂傳說
那天，我揣著相機與錄音利耍，去河邊找鬼蚌本。笑死，一個胖子當著我的面吹牛隘梨，可吹牛的內容都是我干的程癌。我是一名探鬼主播，決...
沈念sama閱讀 40,145評論 3贊 418
雙鴛鴦連環(huán)套：你想象不到人心有多黑
文/蒼蘭香墨我猛地睜開眼轴猎，長吁一口氣：“原來是場噩夢啊……” “哼嵌莉！你這毒婦竟也來了？” 一聲冷哼從身側響起捻脖，我...
開封第一講書人閱讀 39,008評論 0贊 276
萬榮殺人案實錄
序言：老撾萬榮一對情侶失蹤锐峭，失蹤者是張志新（化名）和其女友劉穎，沒想到半個月后可婶，有當?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體沿癞，經...
沈念sama閱讀 45,443評論 1贊 314
?護林員之死
正文獨居荒郊野嶺守林人離奇死亡，尸身上長有42處帶血的膿包…… 初始之章·張勛以下內容為張勛視角年9月15日...
茶點故事閱讀 37,649評論 3贊 334
?白月光啟示錄
正文我和宋清朗相戀三年矛渴，在試婚紗的時候發(fā)現(xiàn)自己被綠了椎扬。大學時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片。...
茶點故事閱讀 39,795評論 1贊 347
活死人
序言：一個原本活蹦亂跳的男人離奇死亡具温，死狀恐怖蚕涤，靈堂內的尸體忽然破棺而出，到底是詐尸還是另有隱情铣猩，我是刑警寧澤揖铜，帶...
沈念sama閱讀 35,501評論 5贊 345
?日本核電站爆炸內幕
正文年R本政府宣布，位于F島的核電站剂习，受9級特大地震影響蛮位，放射性物質發(fā)生泄漏较沪。R本人自食惡果不足惜鳞绕，卻給世界環(huán)境...
茶點故事閱讀 41,119評論 3贊 328
男人毒藥：我在死后第九天來索命
文/蒙蒙一失仁、第九天我趴在偏房一處隱蔽的房頂上張望。院中可真熱鬧们何，春花似錦萄焦、人聲如沸。這莊子的主人今日做“春日...
開封第一講書人閱讀 31,731評論 0贊 22
一樁弒父案拂封，背后竟有這般陰謀
文/蒼蘭香墨我抬頭看了看天上的太陽。三九已至鹦蠕，卻和暖如春冒签，著一層夾襖步出監(jiān)牢的瞬間，已是汗流浹背钟病。一陣腳步聲響...
開封第一講書人閱讀 32,865評論 1贊 269
情欲美人皮
我被黑心中介騙來泰國打工萧恕，沒想到剛下飛機就差點兒被人妖公主榨干…… 1. 我叫王不留，地道東北人肠阱。一個月前我還...
沈念sama閱讀 47,899評論 2贊 370
代替公主和親
正文我出身青樓票唆，卻偏偏與公主長得像，于是被迫代替她去往敵國和親屹徘。傳聞我的和親對象是個殘疾皇子走趋，可洞房花燭夜當晚...
茶點故事閱讀 44,724評論 2贊 354

kafka介紹和使用

推薦閱讀更多精彩內容