2. 程式人生 > >kafka詳解三：開發Kafka應用

kafka詳解三：開發Kafka應用

阿新 • • 發佈：2019-01-11
問題導讀

1.Kafka系統由什麼組成？
2.Kafka中和producer相關的API是什麼？












一、整體看一下Kafka
我們知道，Kafka系統有三大元件：Producer、Consumer、broker 。
 
file:///C:/Users/ADMINI~1/AppData/Local/Temp/enhtmlclip/Image.png
producers 生產(produce)訊息(message)並推(push)送給brokers，consumers從brokers把訊息提取(pull)出來消費(consume)。

二、開發一個Producer應用 
     Producers用來生產訊息並把產生的訊息推送到Kafka的Broker。Producers可以是各種應用，比如web應用，伺服器端應用，代理應用以及log系統等等。當然，Producers現在有各種語言的實現比如Java、C、Python等。

     我們先看一下Producer在Kafka中的角色：

 file:///C:/Users/ADMINI~1/AppData/Local/Temp/enhtmlclip/Image(1).png

2.1.kafka Producer 的 API
Kafka中和producer相關的API有三個類
  • Producer：最主要的類，用來建立和推送訊息
  • KeyedMessage：定義要傳送的訊息物件，比如定義傳送給哪個topic，partition key和傳送的內容等。
  • ProducerConfig:配置Producer，比如定義要連線的brokers、partition class、serializer class、partition key等


2.2下面我們就寫一個最簡單的Producer：產生一條訊息並推送給broker
  1. package bonree.producer;
  2. import java.util.Properties;
  3. import kafka.javaapi.producer.Producer;
  4. import kafka.producer.KeyedMessage;
  5. import kafka.producer.ProducerConfig;
  6. /*******************************************************************************
  7. * BidPlanStructForm.java Created on 2014-7-8
  8. * Author: <a href=mailto:[email protected]>houda</a>
  9. * @Title: SimpleProducer.java
  10. * @Package bonree.producer
  11. * Description:
  12. * Version: 1.0
  13. ******************************************************************************/
  14. public class SimpleProducer {
  15.         private static Producer<Integer,String> producer;
  16.         private final Properties props=new Properties();
  17.         public SimpleProducer(){
  18.                 //定義連線的broker list
  19.                 props.put("metadata.broker.list", "192.168.4.31:9092");
  20.                 //定義序列化類（Java物件傳輸前要序列化）
  21.                 props.put("serializer.class", "kafka.serializer.StringEncoder");
  22.                 producer = new Producer<Integer, String>(new ProducerConfig(props));
  23.         }
  24.         public static void main(String[] args) {
  25.                 SimpleProducer sp=new SimpleProducer();
  26.                 //定義topic
  27.                 String topic="mytopic";
  28.                 //定義要傳送給topic的訊息
  29.                 String messageStr = "send a message to broker ";
  30.                 //構建訊息物件
  31.                 KeyedMessage<Integer, String> data = new KeyedMessage<Integer, String>(topic, messageStr);
  32.                 //推送訊息到broker
  33.                 producer.send(data);
  34.                 producer.close();
  35.         }
  36. }
複製程式碼
三、開發一個consumer應用
     Consumer是用來消費Producer產生的訊息的，當然一個Consumer可以是各種應用，如可以是一個實時的分析系統，也可以是一個數據倉庫或者是一個基於釋出訂閱模式的解決方案等。Consumer端同樣有多種語言的實現，如Java、C、Python等。
     我們看一下Consumer在Kafka中的角色：
 
3.1.kafka Producer 的 API
     Kafka和Producer稍微有些不同，它提供了兩種型別的API
  • high-level consumer API:提供了對底層API的抽象，使用起來比較簡單
  • simple consumer API:允許重寫底層API的實現，提供了更多的控制權，當然使用起來也複雜一些


由於是第一個應用，我們這部分使用high-level API，它的特點每消費一個message自動移動offset值到下一個message，關於offset在後面的部分會單獨介紹。與Producer類似，和Consumer相關的有三個主要的類：
  • KafkaStream：這裡面返回的就是Producer生產的訊息
  • ConsumerConfig：定義要連線zookeeper的一些配置資訊（Kafka通過zookeeper均衡壓力，具體請查閱見面幾篇文章），比如定義zookeeper的URL、group id、連線zookeeper過期時間等。
  • ConsumerConnector:負責和zookeeper進行連線等工作


3.2下面我們就寫一個最簡單的Consumer：從broker中消費一個訊息

  1. package bonree.consumer;
  2. import java.util.HashMap;
  3. import java.util.List;
  4. import java.util.Map;
  5. import java.util.Properties;
  6. import kafka.consumer.Consumer;
  7. import kafka.consumer.ConsumerConfig;
  8. import kafka.consumer.ConsumerIterator;
  9. import kafka.consumer.KafkaStream;
  10. import kafka.javaapi.consumer.ConsumerConnector;
  11. /*******************************************************************************
  12. * Created on 2014-7-8 Author: <a
  13. * href=mailto:[email protected]>houda</a>
  14. * @Title: SimpleHLConsumer.java
  15. * @Package bonree.consumer Description: Version: 1.0
  16. ******************************************************************************/
  17. public class SimpleHLConsumer {
  18.         private final ConsumerConnector consumer;
  19.         private final String topic;
  20.         public SimpleHLConsumer(String zookeeper, String groupId, String topic) {
  21.                 Properties props = new Properties();
  22.                 //定義連線zookeeper資訊
  23.                 props.put("zookeeper.connect", zookeeper);
  24.                 //定義Consumer所有的groupID，關於groupID，後面會繼續介紹
  25.                 props.put("group.id", groupId);
  26.                 props.put("zookeeper.session.timeout.ms", "500");
  27.                 props.put("zookeeper.sync.time.ms", "250");
  28.                 props.put("auto.commit.interval.ms", "1000");
  29.                 consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
  30.                 this.topic = topic;
  31.         }
  32.         public void testConsumer() {
  33.                 Map<String, Integer> topicCount = new HashMap<String, Integer>();
  34.                 //定義訂閱topic數量
  35.                 topicCount.put(topic, new Integer(1));
  36.                 //返回的是所有topic的Map
  37.                 Map<String, List<KafkaStream<byte[], byte[]>>> consumerStreams = consumer.createMessageStreams(topicCount);
  38.                 //取出我們要需要的topic中的訊息流
  39.                 List<KafkaStream<byte[], byte[]>> streams = consumerStreams.get(topic);
  40.                 for (final KafkaStream stream : streams) {
  41.                         ConsumerIterator<byte[], byte[]> consumerIte = stream.iterator();
  42.                         while (consumerIte.hasNext())
  43.                                 System.out.println("Message from Single Topic :: " + new String(consumerIte.next().message()));
  44.                 }
  45.                 if (consumer != null)
  46.                         consumer.shutdown();
  47.         }
  48.         public static void main(String[] args) {
  49.                 String topic = "mytopic";
  50.                 SimpleHLConsumer simpleHLConsumer = new SimpleHLConsumer("192.168.4.32:2181", "testgroup", topic);
  51.                 simpleHLConsumer.testConsumer();
  52.         }
  53. }
複製程式碼
四、執行檢視結果
先啟動伺服器端相關程序：
  • 執行zookeeper：[[email protected] kafka-0.8]# bin/zookeeper-server-start.sh config/zookeeper.properties
  • 執行Kafkabroker：[[email protected] kafka-0.8]# bin/kafka-server-start.sh config/server.properties


再執行我們寫的應用
  • 執行剛才寫的SimpleHLConsumer 類的main函式，等待生產者生產訊息
  • 執行SimpleProducer的main函式，生產訊息並push到broker


結果：執行完SimpleProducer後在SimpleHLConsumer的控制檯即可看到生產者生產的訊息：“send a message to broker”。

轉載： http://www.aboutyun.com/thread-11115-1-1.html

相關推薦

kafka詳解三：開發Kafka應用

問題導讀1.Kafka系統由什麼組成？2.Kafka中和producer相關的API是什麼？一、整體看一下Kafka我們知道，Kafka系統有三大元件：Producer、Consumer、broker 。 file:///C:/Users/ADMINI~1/AppData/

HTTPS協議詳解(三)：PKI 體系

客戶 判斷 節點 無法 三方 證書無效 進行 證書 roo 1、RSA身份驗證的隱患 身份驗證和密鑰協商是TLS的基礎功能，要求的前提是合法的服務器掌握著對應的私鑰。但RSA算法無法確保服務器身份的合法性，因為公鑰並不包含服務器的信息，存在安全隱患: 客戶端C和

mysql系列詳解三：mysql中各類日誌詳解-技術流ken

1.前言 日誌檔案記錄了MySQL資料庫的各種型別的活動，MySQL資料庫中常見的日誌檔案有 查詢日誌，慢查詢日誌，錯誤日誌，二進位制日誌，中繼日誌 。下面分別對他們進行介紹。   2.查詢日誌   1.檢視查詢日誌變數 查詢日誌即檢視日誌記錄了所有對 My

docker容器詳解三：第一個docker程式

新建一個docker 下面來通過命令成功執行一個容器 通過docker images/image ls/docker list/等命令來檢視映象。 命令顯示的分別是: 映象名 標籤(在映象名

java io詳解三：位元組輸入輸出流

這篇部落格我們講的是位元組輸入輸出流：InputStream、OutputSteam(下圖紅色長方形框內)，紅色橢圓框內是其典型實現（FileInputSteam、FileOutStream） 　　    1、位元組輸出流：OutputStream

spice協議詳解(三)：spice架構和通訊

一.相關概念 Channel(通道)： Client和server通過通道進行互動。每個通道型別專用於一種特殊資料型別傳輸。每個通道用一個專用的TCP套接字，可能是安全的（用SSL加密）或不安全的

ES6詳解三：class

class是es6引入的最重要特性之一。在沒有class之前，我們只能通過原型鏈來模擬類。 基本用法 如果你用過java這樣的純面嚮物件語言，那麼你會對class的語法非常熟悉。 class People { constructor(name)

Netty4詳解三：Netty概要講解，全面瞭解Netty，做到胸有成竹

讀完這一章，我們基本上可以瞭解到Netty所有重要的元件，對Netty有一個全面的認識，這對下一步深入學習Netty是十分重要的，而學完這一章，我們其實已經可以用Netty解決一些常規的問題了。一、先縱覽一下Netty，看看Netty都有哪些元件？ 為了更好的理解和進一步深

C++11多執行緒(七):《 詳解三：std::future & std::shared_future》

#include <iostream>                // std::cout #include <future>                // std::async, std::future #include <chrono>            

kafka配置檔案詳解之：server.properties

#每一個broker在叢集中的唯一表示，要求是正數。當該伺服器的IP地址發生改變時，broker.id沒有變化，則不會影響consumers的訊息情況broker.id=0#broker server服務埠 port =9092#處理網路請求的執行緒數量num

kafka配置檔案詳解之：producer.properties

#指定節點列表 metadata.broker.list=kafka01:9092,kafka02:9092,kafka03:9092 #指定分割槽處理類。預設kafka.producer.D

Kafka 詳解（三）------Producer生產者

spl ror 接下來 gif compile inter str lz4 arrays 　　在第一篇博客我們了解到一個kafka系統，通常是生產者Producer 將消息發送到 Broker，然後消費者 Consumer 去 Broker 獲取，那麽本篇博客我們來介紹什麽

大數據新手之路三：安裝Kafka

strac .get 解壓 配置文件 down rac ubuntu 信息 blog WIN7+Ubuntu16.04+Kafka1.0.0 1.下載kafka_2.11-1.0.0.tgz http://kafka.apache.org/downloads

大數據入門第十七天——storm上遊數據源 之kafka詳解（一）入門

不同 這也 接受 blog 存儲 發送 records ant post 一、概述 　　1.kafka是什麽 　　　　根據標題可以有個概念：kafka是storm的上遊數據源之一，也是一對經典的組合，就像郭德綱和於謙 　　　　根據官網：http://kafka.apa

三：python 對象類型詳解一：數字（上）

結果 dom 運行 精度 升級 方法 函數 般的 代碼 一：python 的數字類型： a)整數和浮點數 b)復數 c)固定精度的十進制數 d)有理分數 e)集合 f)布爾類型 g)無窮的整數精度 h)各種數字內置函數和模塊 二：各種數字類型的詳解 　　1，數字常量：pyt

Kafka 溫故(三)：Kafka的內部機制深入(持久化，分散式，通訊協議)

一.Kafka的持久化 1.資料持久化：     發現線性的訪問磁碟（即：按順序的訪問磁碟），很多時候比隨機的記憶體訪問快得多，而且有利於持久化；     傳統的使用記憶體做為磁碟的快取     Kafka直接將資料寫

Kafka 詳解（二）------集群搭建

需要 kafka ast enter 安裝過程 簡單 keep 三臺 sof 　　這裏通過 VMware ，我們安裝了三臺虛擬機，用來搭建 kafka集群，虛擬機網絡地址如下： 　　　hostname ipaddress

Kafka 詳解（二）------叢集搭建

　　這裡通過 VMware ，我們安裝了三臺虛擬機器，用來搭建 kafka叢集，虛擬機器網路地址如下： 　　　hostname                      ipaddress &nbs

【乾貨】一文詳解計算機視覺的廣泛應用：網路壓縮、視覺問答、視覺化、風格遷移等

引言 深度學習目前已成為發展最快、最令人興奮的機器學習領域之一，許多卓有建樹的論文已經發表，而且已有很多高質量的開源深度學習框架可供使用。然而，論文通常非常簡明扼要並假設讀者已對深度學習有相當的理解，這使得初學者經常卡在一些概念的理解上，讀論文似懂非懂，十分吃力。另一方面，即使有了簡單易用的深度

Android Studio 外掛開發詳解二：工具類

轉載請標明出處：http://blog.csdn.net/zhaoyanjun6/article/details/78112856 本文出自【趙彥軍的部落格】 在外掛開發過程中，我們按照開發一個正式的專案來操作，需要整理一些常用工具類。 Http 請求封裝 在外掛的專案中，我們看到依賴庫如下圖所示：