2. 程式人生 > >Kafka應用入門

Kafka應用入門

阿新 • • 發佈：2019-01-10

Kafka是一款基於釋出和訂閱的訊息系統
運作方式如圖:
在這裡插入圖片描述
各個元件的作用:
生產者:將資料依據主題，分割槽發往broker
broker:broker接收來自生產者的訊息，為休息設定偏移量，並將訊息儲存到磁碟
zookeeper:管理各個分散式broker節點，為各個節點提供資料共享，資料一致性，選主服務。(可以看一下Zookeeper和Kafka的關係，為啥Kafka依賴Zookeeper )
消費者:從主題分割槽中讀取資料

Kafka的釋出訂閱體現在，生產者將資料一個一個寫入相應的主題，消費者通過訂閱主題，從主題分割槽中讀取一個一個數據。
下面是主題和分割槽圖:
在這裡插入圖片描述
消費者消費規則:


1.主題的一個分割槽只能供同個消費者群組的一個消費者消費。
如圖所示:
在這裡插入圖片描述在這裡插入圖片描述
2.如果分割槽數小於消費者數量，那麼多出來的消費者啥也不幹。
如圖所示
在這裡插入圖片描述
所以增加分割槽，往消費者群組裡增加消費者是橫向伸縮消費能力的主要方式。

Kafka的安裝
1.zookeeper的安裝和使用
先要裝java1.8，然後下載zookeeper包，解壓
(1)先設定配置檔案
將conf目錄下的zoo_sample.cfg更名為zoo.cfg，預設埠為2181
(2)使用bin/zkServer.sh start 啟動zookeeper
2.kafka安裝
解壓縮包即可，預設埠9092，連線的zookeeper埠2181
啟動kafka
bin/kafka-server-start.sh config/server.properties &
也可以在一臺機子上開啟另外2個節點broker然後再啟動
cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties
bin/kafka-server-start.sh config/server-1.properties &
bin/kafka-server-start.sh config/server-2.properties &

簡單實用Kafka
建立主題，只有一個副本，一個分割槽
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
驗證主題
bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic test
往測試主題上釋出訊息:
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
讀取訊息
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning(舊版本)
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning(新版本2.1)

使用java編寫生產者和消費者
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>com.zte.apt</groupId>
	<artifactId>KafkaTest</artifactId>
	<version>0.0.1-SNAPSHOT</version>

	<name>KafkaTest</name>
	<!-- FIXME change it to the project's website -->
	<url>http://www.example.com</url>

	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
		<maven.compiler.source>1.8</maven.compiler.source>
		<maven.compiler.target>1.8</maven.compiler.target>
	</properties>

	<dependencies>
		<dependency>
			<groupId>org.apache.kafka</groupId>
			<artifactId>kafka-clients</artifactId>
			<version>2.1.0</version>
		</dependency>
		<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.25</version>
</dependency>
		
	</dependencies>

	<build>
		<pluginManagement><!-- lock down plugins versions to avoid using Maven 
				defaults (may be moved to parent pom) -->
			<plugins>
			<plugin>  
            <groupId>org.apache.maven.plugins</groupId>  
            <artifactId>maven-assembly-plugin</artifactId>
            <configuration>
            <archive>  
                    <manifest>  
                        <mainClass>com.zte.apt.KafkaTest.KafkaProducerTest</mainClass>  
                    </manifest>  
                </archive>
                <descriptorRefs>  
                    <descriptorRef>jar-with-dependencies</descriptorRef>  
                </descriptorRefs>  
            </configuration>  
            <executions>  
                <execution>  
                    <id>make-assembly</id>  
                    <phase>package</phase>  
                    <goals>  
                        <goal>single</goal>  
                    </goals>  
                </execution>  
            </executions>  
        </plugin>  
			</plugins>
		</pluginManagement>
	</build>
</project>

生產者
1建立Properties配置物件，配置broker地址，鍵，鍵值的序列化等等
2.建立生產者KafkaProducer
3.建立記錄物件ProducerRecord物件,往ProducerRecord存放主題，鍵，鍵值
4.KafkaProducer傳送ProducerRecord物件，其中傳送方式有:
(1)fire-and-forget傳送並忘記，並不關心是否正常到達，雖然生產者有自動嘗試重發機制，但有時候也會丟失一些資訊
(2)send(ProducerRecord)同步傳送，它會返回一個future物件，呼叫get()方法進行等候，就可以知道訊息是否傳送成功
(3)send(ProducerRecord,Callback)非同步傳送，多一個回撥函式
5.程式碼

package com.zte.KafkaTest;

import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class KafkaProducerTest {
	public static void main(String[] args) {
		Properties props = new Properties();
		props.put("bootstrap.servers", "10.41.41.202:9092");
		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);
		//主題的名字，傳送的鍵，值
		ProducerRecord<String, String> record = new ProducerRecord<>("topictest", "keyhaha", "valuehehe");
		try
		{
			for (int i = 0; i < 100; i++)
			     producer.send(new ProducerRecord<String, String>("topictest", Integer.toString(i), Integer.toString(i)));
			System.out.println("已經發送完畢");
			//非同步傳送，為了傳送成功，主函式最好等待一小段時間
			Thread.sleep(5000);
			
			//同步傳送.呼叫get等待
//			Future<RecordMetadata> result = producer.send(record);
//			System.out.println(result.get().partition());
		}
		catch(Exception e)
		{
			e.printStackTrace();
		}
	}
}

消費者
向消費者群組增加消費者是橫向伸縮消費能力的主要方式，一個主題的每一個分割槽只能分配給同一群組的一個消費者
所以我們有必要為主題建立大量的分割槽，在負載增長時可以加入更多的消費者，不過要注意，不要讓消費者的數量超過主題分割槽的數量，多餘的消費者只會被閒置。
1.建立Properties配置物件，配置broker地址，groupid 群組名字，鍵，鍵值的反序列化等等
2.建立消費者KafkaConsumer
3.訂閱consumer.subscribe(Collections.singletonList(“主題，也可使用正則”))
4.輪詢consumer.poll(100),返回ConsumerRecords記錄物件，
5.提交偏移量，即更新分割槽當前的位置
(1)自動提交設定enable.auto,commit設定為true,每隔一段時間提交一次
(2)同步提交commitSync，阻塞
(3)非同步提交commitAsync，非阻塞
(4)也可以使用非同步和同步結合提交

package com.zte.KafkaTest;

import java.util.Collections;
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;
public class KafkaConsumerTest {

	public static void main(String[] args) 
	{
		Properties props = new Properties();
		props.put("bootstrap.servers", "10.41.41.202:9092");
		//指定消費者所屬群組的名字
		props.put("group.id", "group1");
		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<>(props);
		//訂閱單個主題
		consumer.subscribe(Collections.singletonList("topictest"));
//		//訂閱多個主題
//		consumer.subscribe(Arrays.asList("foo", "bar"));
//		//使用正則訂閱主題
//		consumer.subscribe("test.*");
		try
		{
			while (true) 
			{
		         ConsumerRecords<String, String> records = consumer.poll(100);
		         for (ConsumerRecord<String, String> record : records)
		         {
		             System.out.printf("tppic=%s,partition=%s,offset=%d, key=%s, value =%s%n",record.topic(),record.partition(),record.offset(),record.key(),record.value());
		         }
		     }
		}
		finally
		{
			consumer.close();
		}
	}
}

相關推薦

Kafka應用入門

Kafka是一款基於釋出和訂閱的訊息系統 運作方式如圖: 各個元件的作用: 生產者:將資料依據主題，分割槽發往broker broker:broker接收來自生產者的訊息，為休息設定偏移量，並將訊息儲存到磁碟 zookeeper:管理各個分散式broker節點，為各個節點提供資料共享，

Kafka系列 —— 入門及應用場景 & 部署 & 簡單測試

Kafka系列為自己學習與使用Kafka中遇到的問題與總結。本系列將介紹如下內容： Kafka入門及應用場景 & 部署 & 簡單測試 Kafka核心概念 Kafka常用命令 Kafka監控 Kafka消費語義分析 Flume + Kafka

kafka經典入門

stat 緩存 大於 分區 依然 fig tin 發送消息 ray 問題導讀1.Kafka獨特設計在什麽地方？2.Kafka如何搭建及創建topic、發送消息、消費消息？3.如何書寫Kafka程序？4.數據傳輸的事務定義有哪三種？5.Kafka判斷一個節點是否活著有哪

Storm集成Kafka應用的開發

集群 with end putc script apache out println dto 　　我們知道storm的作用主要是進行流式計算，對於源源不斷的均勻數據流流入處理是非常有效的，而現實生活中大部分場景並不是均勻的數據流，而是時而多時而少的數據流入，這種情況下顯然用

二分的幾種簡單應用——入門

答案 ... 繼續 med 單位 nes not to do 條件 ---恢復內容開始--- 說起二分，最基礎的二分搜索算法就不用了吧（最難的還際應用啊感覺）……實現起來很簡單，搜索區間不斷減半……唔……嘛……簡單甩個模板好了(●‘?‘●) 下面開始學習二分的幾種應用啦~

windows下kafka配置入門 示例

keep bsp col fine program summer actor 回車 打開 實驗平臺與軟件： 操作系統：windows7 32 位 java 開發包: jdk1.8.0_144 集群: zookeeper-3.3.6 消息隊列: kafka_2.11

kubernetes應用入門

指定 let 對象 mes 8.0 ast res svc names 一、kubernetes資源對象 pod serivce replicaset deployment statefulet daemonset job cronjob node 二、kubectl命

簡單快取及快取表應用入門

業務需求：源資料庫比較龐大，一個簡單查詢都要幾秒到十幾秒。現在需要統計某日的部分資料，比如當天有多少使用者登入之類，需要按天統計，還需要有個功能要合計。如果一天天去查詢，查一次耗費的時間都很長，不用快取的話，下次再查需要消耗更長時間。 分析與解決：除了當天外，任何一天的資料統計其實都是穩定不變的，

Storm整合Kafka應用的開發

https://www.cnblogs.com/freeweb/p/5292961.html Storm整合Kafka應用的開發 　　我們知道storm的作用主要是進行流式計算，對於源源不斷的均勻資料流流入處理是非常有效的，而現實生活中大部分場景並不是均勻的資料流，而是時而多時而少的資料流入

快應用入門：第一個快應用程式

簡介 快應用是九大手機廠商基於硬體平臺共同推出的新型應用生態，依附在各大應用市場。使用者無需下載安裝，即點即用，享受原生應用的效能體驗。2018年3月20日在北京推出“快應用”標準。該應用模組以標準js語法+重組css標籤+Flexbox佈局+mvvm模式。 第一

從零開始——PowerShell應用入門

薰薰熊的技術分享社群   部落格園 首頁 新隨筆 聯絡 訂閱 管理 隨筆 - 18  文章 - 0  評論 - 42 從零開始——PowerShell應用入門（全例子入門講解） 學習一門技術，不止要

Kafka簡單入門與Spring結合實踐

Kafka簡單入門與Spring結合實踐 一、【安裝部署kafka伺服器環境（centos7.0）】:  1.【注意】新版的kafka已經內建了一個zookeeper環境 2.【安裝與執行】： 可以在kafka官網 http://kafka.ap

Kafka架構、Kafka核心元件、Kafka工作原理、Kafka應用場景

什麼是訊息系統？          訊息系統負責將資料從一個應用程式傳輸到另一個應用程式，因此應用程式可以專注於資料，但不擔心如何共享它。 分散式訊息傳遞基於可靠訊息佇列的概念。 訊息在客戶端應用程式和訊息傳遞系統之間非同步排隊。 有兩種型別的訊

一、Kafka概念入門

Kafka是分散式釋出-訂閱訊息系統。它最初由LinkedIn公司開發，之後成為Apache專案的一部分。Kafka是一個分散式的，可劃分的，冗餘備份的永續性的日誌服務。它主要用於處理活躍的流式資料。 在大資料系統中，常常會碰到一個問題，整個大資料是由各個子系統組成，資料需要在各個子系統中高效能

WebSocket 的簡單應用入門

websocket簡單實現分為以下幾個步驟：新增websocket庫、編寫後臺程式碼、編寫前端程式碼。 新增websocket庫 在maven中新增websocket庫的程式碼如下所示： <dependency> <groupId>ja

solr應用入門例項

教程網址：https://blog.csdn.net/liaomin416100569/article/details/77301756 一 、solr簡介        solr是以lucene為核心開發的企業級

Kafka從入門到精通 第二章 kafka的安裝

一  環境準備 1.1 叢集規劃 這裡需要三個節點，分別如下： linux102                    linux103                linux104 zk                            zk          

git簡單應用入門--從實際專案開發出發

針對第一次git的大白，結合自己實際專案開發，梳理了以下內容，希望對初學者有幫助。 1. 拉取遠端指定分支，在指定分支上開發 首先，預設遠端有兩個分支master和dev 1.1 拉取遠端master分支，同時新建本地master分支，本地master分支與遠

使用SpringBoot開發JavaWeb應用入門

使用SpringBoot開發JavaWeb應用入門 0、環境及開發工具準備 環境 開發工具 1、建立專案 統一編碼 建立SpringBoot專案 啟動看看效果 2、編寫控制器(Con

java B2B2C原始碼電子商城系統-Kafka快速入門

大家對Kafka有了一些基本瞭解之後，下面我們來嘗試構建一個Kafka服務端，並體驗一下基於Kafka的訊息生產與消費。 需要JAVA Spring Cloud大型企業分散式微服務雲構建的B2B2C電子商務平臺原始碼 一零三八七七四六二六 環境安裝 首先，我們需要從官網上下載安裝介質。 在解