真实的国产乱ⅩXXX66竹夫人,五月香六月婷婷激情综合,亚洲日本VA一区二区三区,亚洲精品一区二区三区麻豆

成都創(chuàng)新互聯(lián)網站制作重慶分公司

Kafka怎么利用Java實現(xiàn)數(shù)據(jù)的生產和消費實例教程

這篇文章主要介紹Kafka怎么利用Java實現(xiàn)數(shù)據(jù)的生產和消費實例教程，文中介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們一定要看完！

只為您設計更接底氣、較有營銷力的好網站，將營銷策劃與網頁設計互相結合的專業(yè)機構，營銷型網站公司中較早掌握H5頁面制作技術的機構。一個好的成都品牌網站建設，不能只是一張名片，茫茫網海，想要快速吸引到您客戶的眼球，必須全方位的展現(xiàn)出企業(yè)突出的優(yōu)勢，以求達到主動營銷的效果，最終促成成交！

Kafka的介紹

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，它可以處理消費者規(guī)模的網站中的所有動作流數(shù)據(jù)。

Kafka 有如下特性：

  • 以時間復雜度為O(1)的方式提供消息持久化能力，即使對TB級以上數(shù)據(jù)也能保證常數(shù)時間復雜度的訪問性能。

  • 高吞吐率。即使在非常廉價的商用機器上也能做到單機支持每秒100K條以上消息的傳輸。

  • 支持Kafka Server間的消息分區(qū)，及分布式消費，同時保證每個Partition內的消息順序傳輸。

  • 同時支持離線數(shù)據(jù)處理和實時數(shù)據(jù)處理。

  • Scale out：支持在線水平擴展。

kafka的術語

  • Broker：Kafka集群包含一個或多個服務器，這種服務器被稱為broker。

  • Topic：每條發(fā)布到Kafka集群的消息都有一個類別，這個類別被稱為Topic。（物理上不同Topic的消息分開存儲，邏輯上一個Topic的消息雖然保存于一個或多個broker上但用戶只需指定消息的Topic即可生產或消費數(shù)據(jù)而不必關心數(shù)據(jù)存于何處）

  • Partition：Partition是物理上的概念，每個Topic包含一個或多個Partition。

  • Producer：負責發(fā)布消息到Kafka broker。

  • Consumer：消息消費者，向Kafka broker讀取消息的客戶端。

  • Consumer Group:每個Consumer屬于一個特定的Consumer Group（可為每個Consumer指定group name，若不指定group name則屬于默認的group）。

kafka核心Api

kafka有四個核心API

  • 應用程序使用producer API發(fā)布消息到1個或多個topic中。

  • 應用程序使用consumer API來訂閱一個或多個topic,并處理產生的消息。

  • 應用程序使用streams API充當一個流處理器,從1個或多個topic消費輸入流,并產生一個輸出流到1個或多個topic,有效地將輸入流轉換到輸出流。

  • connector API允許構建或運行可重復使用的生產者或消費者,將topic鏈接到現(xiàn)有的應用程序或數(shù)據(jù)系統(tǒng)。

示例圖如下:

Kafka怎么利用Java實現(xiàn)數(shù)據(jù)的生產和消費實例教程

kafka 應用場景

  • 構建可在系統(tǒng)或應用程序之間可靠獲取數(shù)據(jù)的實時流數(shù)據(jù)管道。

  • 構建實時流應用程序，可以轉換或響應數(shù)據(jù)流。

以上介紹參考kafka官方文檔。

開發(fā)準備

如果我們要開發(fā)一個kafka的程序，應該做些什么呢？

首先，在搭建好kafka環(huán)境之后，我們要考慮的是我們是生產者還是消費者，也就是消息的發(fā)送者還是接受者。
不過在本篇中，生產者和消費者都會進行開發(fā)和講解。

在大致的了解kafka之后，我們來開發(fā)第一個程序。

這里用的開發(fā)語言是Java，構建工具Maven。

Maven的依賴如下:

 
  org.apache.kafka
   kafka_2.12
   1.0.0
   provided 
  
  
  
    org.apache.kafka
    kafka-clients
    1.0.0
  
  
  
   org.apache.kafka
   kafka-streams
   1.0.0
  

Kafka Producer

在開發(fā)生產的時候，先簡單的介紹下kafka各種配置說明：

  • bootstrap.servers： kafka的地址。

  • acks:消息的確認機制，默認值是0。

  • acks=0：如果設置為0，生產者不會等待kafka的響應。

  • acks=1：這個配置意味著kafka會把這條消息寫到本地日志文件中，但是不會等待集群中其他機器的成功響應。

  • acks=all：這個配置意味著leader會等待所有的follower同步完成。這個確保消息不會丟失，除非kafka集群中所有機器掛掉。這是最強的可用性保證。

  • retries：配置為大于0的值的話，客戶端會在消息發(fā)送失敗時重新發(fā)送。

  • batch.size:當多條消息需要發(fā)送到同一個分區(qū)時，生產者會嘗試合并網絡請求。這會提高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 producer = new KafkaProducer(props);

kafka的配置添加之后，我們便開始生產數(shù)據(jù)，生產數(shù)據(jù)代碼只需如下就行:

producer.send(new ProducerRecord(topic,key,value));

  • topic: 消息隊列的名稱，可以先行在kafka服務中進行創(chuàng)建。如果kafka中并未創(chuàng)建該topic，那么便會自動創(chuàng)建！

  • key:鍵值，也就是value對應的值，和Map類似。

  • value:要發(fā)送的數(shù)據(jù)，數(shù)據(jù)格式為String類型的。

在寫好生產者程序之后，那我們先來生產吧！

我這里發(fā)送的消息為:

 String messageStr="你好，這是第"+messageNo+"條數(shù)據(jù)";

并且只發(fā)送1000條就退出，結果如下:

Kafka怎么利用Java實現(xiàn)數(shù)據(jù)的生產和消費實例教程

可以看到信息成功的打印了。

如果不想用程序進行驗證程序是否發(fā)送成功，以及消息發(fā)送的準確性，可以在kafka服務器上使用命令查看。

Kafka Consumer

kafka消費這塊應該來說是重點，畢竟大部分的時候，我們主要使用的是將數(shù)據(jù)進行消費。

kafka消費的配置如下:

  • bootstrap.servers： kafka的地址。

  • group.id：組名 不同組名可以重復消費。例如你先使用了組名A消費了kafka的1000條數(shù)據(jù)，但是你還想再次進行消費這1000條數(shù)據(jù)，并且不想重新去產生，那么這里你只需要更改組名就可以重復消費了。

  • enable.auto.commit：是否自動提交，默認為true。

  • auto.commit.interval.ms: 從poll(拉)的回話處理時長。

  • session.timeout.ms:超時時間。

  • max.poll.records:一次最大拉取的條數(shù)。

  • auto.offset.reset：消費規(guī)則，默認earliest 。
     earliest: 當各分區(qū)下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費 。
     latest: 當各分區(qū)下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該分區(qū)下的數(shù)據(jù) 。
     none: topic各分區(qū)都存在已提交的offset時，從offset后開始消費；只要有一個分區(qū)不存在已提交的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 consumer = new KafkaConsumer(props);

由于我這是設置的自動提交，所以消費代碼如下:

我們需要先訂閱一個topic，也就是指定消費哪一個topic。

consumer.subscribe(Arrays.asList(topic));

訂閱之后，我們再從kafka中拉取數(shù)據(jù):

ConsumerRecords msgList=consumer.poll(1000);

一般來說進行消費會使用監(jiān)聽，這里我們就用for(;;)來進行監(jiān)聽， 并且設置消費1000條就退出！

結果如下:

Kafka怎么利用Java實現(xiàn)數(shù)據(jù)的生產和消費實例教程

可以看到我們這里已經成功消費了生產的數(shù)據(jù)了。

代碼

那么生產者和消費者的代碼如下:

生產者:

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 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(props);
  this.topic = topicName;
 }

 @Override
 public void run() {
  int messageNo = 1;
  try {
   for(;;) {
    String messageStr="你好，這是第"+messageNo+"條數(shù)據(jù)";
    producer.send(new ProducerRecord(topic, "Message", messageStr));
    //生產了100條就打印
    if(messageNo%100==0){
     System.out.println("發(fā)送的信息:" + messageStr);
    }
    //生產1000條就退出
    if(messageNo%1000==0){
     System.out.println("成功發(fā)送了"+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 consumer;
 private ConsumerRecords 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(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 record : msgList) {
      //消費100條就打印 ,但打印的數(shù)據(jù)不一定是這個規(guī)律的
      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 是因為我在自己的環(huán)境做了關系映射，這個可以換成服務器的IP。

當然項目我放在Github上了，有興趣的可以看看。 https://github.com/xuwujing/kafka （本地下載）

總結

簡單的開發(fā)一個kafka的程序需要以下步驟:

  • 成功搭建kafka服務器，并成功啟動！

  • 得到kafka服務信息，然后在代碼中進行相應的配置。

  • 配置完成之后，監(jiān)聽kafka中的消息隊列是否有消息產生。

  • 將產生的數(shù)據(jù)進行業(yè)務邏輯處理！

以上是“Kafka怎么利用Java實現(xiàn)數(shù)據(jù)的生產和消費實例教程”這篇文章的所有內容，感謝各位的閱讀！希望分享的內容對大家有幫助，更多相關知識，歡迎關注創(chuàng)新互聯(lián)行業(yè)資訊頻道！


本文名稱：Kafka怎么利用Java實現(xiàn)數(shù)據(jù)的生產和消費實例教程
文章起源：http://weahome.cn/article/ijjssc.html

其他資訊

在線咨詢

微信咨詢

電話咨詢

028-86922220（工作日）

18980820575（7×24）

提交需求

返回頂部