2. 程式人生 > >Kafka 使用Java實現資料的生產和消費demo

Kafka 使用Java實現資料的生產和消費demo

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

前言

在上一篇中講述如何搭建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的依賴如下:

    <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配置如下:
        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));
  • topic: 訊息佇列的名稱，可以先行在kafka服務中進行建立。如果kafka中並未建立該topic，那麼便會自動建立！
  • key:鍵值，也就是value對應的值，和Map類似。

  • value:要傳送的資料，資料格式為String型別的。

    在寫好生產者程式之後，那我們先來生產吧！
    我這裡傳送的訊息為:

 String messageStr="你好，這是第"+messageNo+"條資料";

並且只發送1000條就退出，結果如下:
這裡寫圖片描述

可以看到資訊成功的列印了。
如果不想用程式進行驗證程式是否傳送成功，以及訊息傳送的準確性，可以在kafka伺服器上使用命令檢視。

Kafka Consumer

kafka消費這塊應該來說是重點，畢竟大部分的時候，我們主要使用的是將資料進行消費。

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配置如下:

    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。

總結

簡單的開發一個kafka的程式需要以下步驟:

  1. 成功搭建kafka伺服器，併成功啟動！
  2. 得到kafka服務資訊，然後在程式碼中進行相應的配置。
  3. 配置完成之後，監聽kafka中的訊息佇列是否有訊息產生。
  4. 將產生的資料進行業務邏輯處理！

到此，本文就結束了，謝謝閱讀！

相關推薦

Kafka 使用Java實現資料的生產和消費demo

前言 在上一篇中講述如何搭建kafka叢集，本篇則講述如何簡單的使用 kafka 。不過在使用kafka的時候，還是應該簡單的瞭解下kafka。 Kafka的介紹 Kafka是一種高吞吐量的分散式釋出訂閱訊息系統，它可以處理消費者規模的網站中的所有動作流資

Kafka 使用Java實現數據的生產和消費demo

== 輸入流 是你 www esc 驗證 rri else 分開 前言 在上一篇中講述如何搭建kafka集群，本篇則講述如何簡單的使用 kafka 。不過在使用kafka的時候，還是應該簡單的了解下kafka。 Kafka的介紹 Kafka是一種高吞吐量的分布式發布訂閱消息

部署kafka+zookeeper叢集，java實現生產和消費

很長一段時間沒有寫部落格了。最近單位系統升級，鑑於activemq+redis構架存在的一些我沒有能力解決的問題。考慮再三，決定換kafka+zookeeper。選用的是多節點多boker叢集。 先說一下我的伺服器歡迎： 編號 伺服器名 內網IP 0

Spring-Kafka XML配置方法實現生產和消費

  Spring-Kafka XML配置方法實現生產和消費  1. 生產者配置xml Step1：producerProperties：設定生產者公產需要的配置 ； Step2：producerFactory：定義了生產者工廠構造方法

c語言使用librdkafka庫實現kafka的生產和消費例項

關於librdkafka庫的介紹，可以參考kafka的c/c++高效能客戶端librdkafka簡介，本文使用librdkafka庫來進行kafka的簡單的生產、消費 一、producer librdkafka進行kafka生產操作的大致步驟如下： 1、建立kafka

安裝activemq和java代碼實現生產和消費

uri 技術分享 啟動 開發 保存 realm 不下來 選擇 jetty Apache Active MQ的官網 http://activemq.apache.org/ 下載地址： 國外的網站，網速不好的時候，真的下不下來 http://activemq.apache

關於Kafka遠端生產和消費的額外說明

網上有太多此類帖子，然而本帖比他們全面。 背景 有一臺阿里雲伺服器A，用作測試伺服器，一臺華為伺服器B，用作生產伺服器。均搭建相同配置和版本Apache Kafka。現在需遠端生產和消費Kafka。在伺服器B均無錯誤，在伺服器A死活不通。 問題排查 iptable

kafka java 實現訊息佇列demo

kafka是吞吐量巨大的一個訊息系統，它是用scala寫的，和普通的訊息的生產消費還有所不同，寫了個demo程式供大家參考。kafka的安裝請參考官方文件。首先我們需要新建一個maven專案，然後在pom中引用kafka jar包，引用依賴如下： <depend

Java 基礎資料型別 和 深度克隆物件的2種方法(實現Cloneable介面或者實現物件序列化)

知識點1(java有9種傳遞函式引數時是傳值): 8大基礎資料型別加String: 1,位元組 byte   2,短整型 short  3,整型int  4,長整形 long  5,字元型 char  6,浮點型 float  7,雙精度型 double  8,布林型 bo

Java實現對zip和rar文件的解壓縮

資源 types num tar rec absolut cfi pub more package com.svse.test; import java.io.File;import java.io.FileOutputStream;import java.io.IOExc

Java實現資料統計的常用演算法

求和、平均值、眾數、中位數、中列數、四分位數、極差、四分位數、截斷均值、方差、絕對平均差（AAD）、中位數絕對偏差、標準差 的數學方法 package cn.javacodes.utils; import java.util.Arrays; import java.util.HashMap;

Java byte資料轉換和處理總結

一.byte和int相互轉換的方法 java程式或Android程式的socket資料傳輸，都是通過byte陣列，但是int型別是4個byte組成的，如何把一個整形int轉換成byte陣列，同時如何把一個長度為4的byte陣列轉換為int型別。 /** * int到byte[]

java面試-資料結構和演算法

1.排序 1.1 氣泡排序 package sort; /** * Created by david on 2018/8/16 * 氣泡排序 */ public class BubbleSort { private static int[] bubbleSort(int

java實現物件拷貝和屬性複製的使用

java實現物件拷貝和屬性複製的使用   對於某一屬性，即原始物件和目的物件的屬性名稱相同，就可以拷貝原始物件的屬性值到目的物件中。   注意屬性必須新增set，get方法，否則拷貝不成功   基本型別之間如果是屬性的型別不同，BeanUtils會嘗試去強制

Java的資料型別 和自動裝箱拆箱機制

Java的資料型別 和裝箱拆箱機制 資料型別 第一類整數型（byte short int long） Java各整數型有固定的表示範圍和欄位長度 ，不會受到不同作業系統的影響，這點確保了java程式的可移植性。 Java中整數型的資料預設值為int. 表示方式 十進位制：123 八

Java實現資料夾的複製

Java複製當個檔案很簡答，我們直接通過io操作即可。但複製一個資料夾時就不同了。最主要的難點在於不知道檔案的結構。 實現思路： 1.遍歷目標檔案下的所有子檔案和目錄。 2.遍歷所有子檔案判斷是目錄還是檔案，如果是目錄的話先建立該目錄，然後再遞迴進入該目錄遍歷，反之則複製檔案。 &n

java實現Brute-Force和KMP模式匹配

Brute-Force模式匹配演算法 從主串第start（i=start）個字元起，與模式串t的第一個字元（j=0）開始比較。 若相等，則繼續比較後面字元（i++，j++） 若不相等，則從主串第二個字元起重新和模式串t比較（i=i-j+1，j=0） 若都匹配成功，則返回模式串t第

用java實現簡易加密和解密

實現原理：               通過位運算的異或運算子“^”把字串與一個指定的值進行異或運算，從而改變字串中的每個字元的值，這樣就可以得到一個加密後的字串。傳送後再將加密後的字串通過源程式實現解

SpringBoot整合RabbitMQ，實現訊息傳送和消費

下載安裝Erlang和RabbitMQ Erlang和RabbitMQ：https://www.cnblogs.com/theRhyme/p/10069611.html   專案建立和依賴 推薦SpringCloud專案線上建立：https://start.spring.io/ 不用上面這

java基本資料型別和引用型別以及他們的儲存方式

1、基本資料型別： byte 8位 short 16 位 int 32 位 long 64位 float 32位 double 64位 2、基本資料型別和引用資料型別的儲存位置： java中的基本資料型別是儲存在棧中還是儲存在堆中取決於基本型別