2. 程式人生 > >分散式釋出訂閱訊息系統Kafka

分散式釋出訂閱訊息系統Kafka

阿新 • • 發佈：2018-12-09

文章目錄

Kafka概述

卡夫卡是用於構建實時資料管道和流媒體應用。它是水平可伸縮的，容錯的，快速的，並執行在數千家公司的生產。

  • PUBLISH & SUBSCRIBE
    像訊息傳遞系統一樣讀寫資料流。
  • PROCESS
    編寫可伸縮的流式處理應用程式，實時響應事件。
  • STORE
    儲存資料流安全分佈，複製。容錯叢集。

Kafka概述和訊息系統類似

訊息中介軟體：生產者和消費者

媽媽：生產者
你：消費者
饅頭：資料流、訊息

	正常情況下： 生產一個  消費一個
	其他情況：  
		一直生產，你吃到某一個饅頭時，你卡主(機器故障)， 饅頭就丟失了
		一直生產，做饅頭速度快，你吃來不及，饅頭也就丟失了

	拿個碗/籃子，饅頭做好以後先放到籃子裡，你要吃的時候去籃子裡面取出來吃

籃子/框： Kafka
	當籃子滿了，饅頭就裝不下了，咋辦？ 
	多準備幾個籃子 === Kafka的擴容

流處理平臺有三個關鍵功能:

  • 釋出和訂閱資料流，類似於訊息佇列或企業訊息傳遞系統。
  • 以容錯的持久方式儲存記錄流。
  • 處理實時產生的資料流。

Kafka架構及核心概念

在這裡插入圖片描述
producer：生產者，就是生產饅頭(老媽)
consumer：消費者，就是吃饅頭的(你)
broker：籃子
topic：主題，給饅頭帶一個標籤，topica的饅頭是給你吃的，topicb的饅頭是給你弟弟吃

首先是幾個概念:

  • Kafka作為一個叢集執行在一個或多個伺服器上，這些伺服器可以跨多個數據中心。
  • Kafka叢集將記錄流儲存在稱為topic的類別中。
  • 每個記錄由一個key、一個value和一個 timestamp組成。

Kafka單節點單Broker部署之Zookeeper安裝

kafka部署方式：

  • 單節點單Broker部署及使用
  • 單節點多Broker部署及使用
  • 多節點多Broker部署及使用

單節點單Broker部署

1、zk直接解壓；配置環境變數（vi ~/.bash_profile）
在這裡插入圖片描述
修改ZK_HOME下的conf下的zoo.cfg
資料儲存路徑
2、下載解壓kafka；配置環境變數
在這裡插入圖片描述
修改配置檔案-$KAFKA_HOME/config/server.properties

broker.id=0
listeners		//監聽的埠
host.name		//啟動在本臺機器ip
log.dirs		//日誌儲存路徑
zookeeper.connect		//外部zk

啟動Kafka

kafka-server-start.sh

USAGE: /home/hadoop/app/kafka_2.11-0.9.0.0/bin/kafka-server-start.sh [-daemon] server.properties [--override property=value]*

kafka-server-start.sh $KAFKA_HOME/config/server.properties

在這裡插入圖片描述

建立topic: zk

kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 1 --partitions 1 --topic hello_topic

檢視所有topic

kafka-topics.sh --list --zookeeper hadoop000:2181

在這裡插入圖片描述
注意要分別在不同的控制檯執行

傳送訊息: broker

kafka-console-producer.sh --broker-list hadoop000:9092 --topic hello_topic

消費訊息: zk

kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic hello_topic --from-beginning
--from-beginning的使用是從頭消費

檢視所有topic的詳細資訊：kafka-topics.sh --describe --zookeeper hadoop000:2181
檢視指定topic的詳細資訊：kafka-topics.sh --describe --zookeeper hadoop000:2181 --topic hello_topic

在這裡插入圖片描述

單節點多Broker部署及使用

在這裡插入圖片描述
分別修改三個配置檔案

server-1.properties
	log.dirs=/home/hadoop/app/tmp/kafka-logs-1
	listeners=PLAINTEXT://:9093
	broker.id=1

server-2.properties
	log.dirs=/home/hadoop/app/tmp/kafka-logs-2
	listeners=PLAINTEXT://:9094
	broker.id=2

server-3.properties
	log.dirs=/home/hadoop/app/tmp/kafka-logs-3
	listeners=PLAINTEXT://:9095
	broker.id=3

啟動kafak

kafka-server-start.sh -daemon $KAFKA_HOME/config/server-1.properties &
kafka-server-start.sh -daemon $KAFKA_HOME/config/server-2.properties &
kafka-server-start.sh -daemon $KAFKA_HOME/config/server-3.properties &

測試：

//建立一個topic
kafka-topics.sh --create --zookeeper hadoop000:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic
//生產端傳送訊息 
kafka-console-producer.sh --broker-list hadoop000:9093,hadoop000:9094,hadoop000:9095 --topic my-replicated-topic
//消費者消費訊息
kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic my-replicated-topic
//檢視topic的訊息
kafka-topics.sh --describe --zookeeper hadoop000:2181 --topic my-replicated-topic

在這裡插入圖片描述

Kafka容錯性測試

在這裡插入圖片描述
kill掉server2
在這裡插入圖片描述
經過測試消費者和生產者都可以正常進行
查詢topic的詳細資訊發現活得節點就只剩3和1
在這裡插入圖片描述
繼續kill掉server3
在這裡插入圖片描述
測試消費者和生產者都可以正常進行；雖然有warning日誌；但是沒有影響幹活

在scala-maven的IDEA下Kafka Java API程式設計測試

加入kafka依賴

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka_2.11</artifactId>
            <version>0.9.0.0</version>
        </dependency>

/**
 * Kafka常用配置檔案
 */
public class KafkaProperties {

    public static final String ZK = "192.168.199.111:2181";

    public static final String TOPIC = "hello_topic";

    public static final String BROKER_LIST = "192.168.199.111:9092";

    public static final String GROUP_ID = "test_group1";

}

Kafka Producer

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

/**
 * Kafka生產者
 */
public class KafkaProducer extends Thread{

    private String topic;

    private Producer<Integer, String> producer;

    public KafkaProducer(String topic) {
        this.topic = topic;

        Properties properties = new Properties();

        properties.put("metadata.broker.list",KafkaProperties.BROKER_LIST);
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        properties.put("request.required.acks","1");

        producer = new Producer<Integer, String>(new ProducerConfig(properties));
    }


    @Override
    public void run() {

        int messageNo = 1;

        while(true) {
            String message = "message_" + messageNo;
            producer.send(new KeyedMessage<Integer, String>(topic, message));
            System.out.println("Sent: " + message);

            messageNo ++ ;

            try{
                Thread.sleep(2000);
            } catch (Exception e){
                e.printStackTrace();
            }
        }

    }
}

Kafka Consumer

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * Kafka消費者
 */
public class KafkaConsumer extends Thread{

    private String topic;

    public KafkaConsumer(String topic) {
        this.topic = topic;
    }


    private ConsumerConnector createConnector(){
        Properties properties = new Properties();
        properties.put("zookeeper.connect", KafkaProperties.ZK);
        properties.put("group.id",KafkaProperties.GROUP_ID);
        return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));
    }

    @Override
    public void run() {
        ConsumerConnector consumer = createConnector();

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, 1);
//        topicCountMap.put(topic2, 1);
//        topicCountMap.put(topic3, 1);

        // String: topic
        // List<KafkaStream<byte[], byte[]>>  對應的資料流
        Map<String, List<KafkaStream<byte[], byte[]>>> messageStream =  consumer.createMessageStreams(topicCountMap);

        KafkaStream<byte[], byte[]> stream = messageStream.get(topic).get(0);   //獲取我們每次接收到的資料

        ConsumerIterator<byte[], byte[]> iterator = stream.iterator();


        while (iterator.hasNext()) {
            String message = new String(iterator.next().message());
            System.out.println("rec: " + message);
        }
    }
}


/**
 * Kafka Java API測試
 */
public class KafkaClientApp {

    public static void main(String[] args) {
        new KafkaProducer(KafkaProperties.TOPIC).start();

        new KafkaConsumer(KafkaProperties.TOPIC).start();

    }
}

IDEA控制檯
在這裡插入圖片描述
虛擬機器Consumer控制檯
在這裡插入圖片描述
在這裡插入圖片描述

Kafka實戰之整合Flume和Kafka完成實時資料採集

在這裡插入圖片描述
更改Flume實戰案例三中的下半部分avro-memory-logger.conf為avro-memory-kafka.conf

avro-memory-kafka.conf

avro-memory-kafka.sources = avro-source
avro-memory-kafka.sinks = kafka-sink
avro-memory-kafka.channels = memory-channel

avro-memory-kafka.sources.avro-source.type = avro
avro-memory-kafka.sources.avro-source.bind = hadoop000
avro-memory-kafka.sources.avro-source.port = 44444

avro-memory-kafka.sinks.kafka-sink.type = org.apache.flume.sink.kafka.KafkaSink
avro-memory-kafka.sinks.kafka-sink.brokerList = hadoop000:9092
avro-memory-kafka.sinks.kafka-sink.topic = hello_topic
avro-memory-kafka.sinks.kafka-sink.batchSize = 5
avro-memory-kafka.sinks.kafka-sink.requiredAcks =1 

avro-memory-kafka.channels.memory-channel.type = memory

avro-memory-kafka.sources.avro-source.channels = memory-channel
avro-memory-kafka.sinks.kafka-sink.channel = memory-channel

啟動測試：


flume-ng agent \
--name avro-memory-kafka  \
--conf $FLUME_HOME/conf  \
--conf-file $FLUME_HOME/conf/avro-memory-kafka.conf \
-Dflume.root.logger=INFO,console


flume-ng agent \
--name exec-memory-avro  \
--conf $FLUME_HOME/conf  \
--conf-file $FLUME_HOME/conf/exec-memory-avro.conf \
-Dflume.root.logger=INFO,console


kafka-console-consumer.sh --zookeeper hadoop000:2181 --topic hello_topic

相關推薦

分散式釋出訂閱訊息系統Kafka

文章目錄 Kafka概述 Kafka架構及核心概念 Kafka單節點單Broker部署之Zookeeper安裝 單節點單Broker部署 單節點多Broker部署及使用 Kafka容錯性測試

分散式釋出訂閱訊息系統 Kafka

kafka是一種高吞吐量的分散式釋出訂閱訊息系統，她有如下特性： 通過O(1)的磁碟資料結構提供訊息的持久化，這種結構對於即使數以TB的訊息儲存也能夠保持長時間的穩定效能。 高吞吐量：即使是非常普通的硬體kafka也可以支援每秒數十萬的訊息。 支援通過kafka伺服器

大資料系列之分散式釋出訂閱訊息系統Kafka（一）Kafka簡介，組成，叢集安裝

1.Kafka簡介   Kafka如同JMS(Java Message Service)一樣，是一箇中間件，在異構系統間通訊，為不同的系統之間提供服務。我們知道JMS通過佇列（一對一）與主題（一對多）兩種形式提供服務，而Kafka則通過主題（topic），來給一組消費者提供

大資料系列之分散式釋出訂閱訊息系統Kafka（四）Kafka與Flume的3種整合

前面我們已經介紹了Flume，現在我們將Kafka與Flume整合 先看一下Flume的結構組成：            我們可以發現，將Flume與Kafka進行整合無非3種情況，Flume作為生產者——Sink輸出到Kafka，Flume作為消費者——Source接

分散式釋出訂閱訊息系統 Kafka 架構設計

底層API class SimpleConsumer { /* Send fetch request to a broker and get back a set of messages. */ public ByteBufferMessageSet fetch(FetchRequest re

Kafka-API中介軟體MQ訊息佇列在Maven專案中的配置使用操作 (分散式釋出訂閱訊息系統)

一、 Maven依賴 <dependency> <groupId>com.foriseland.fjf.mq</groupId> <artifactId>fjf-mq-kafka</artifactId> &

Kafka（分散式釋出訂閱訊息系統）

http://kafka.apache.org/目前越來越多的開源分散式處理系統如Apache Storm、Spark都支援與Kafka整合。 使用場景：設想這樣一個情景：想分析使用者在網站上的的瀏覽行為。這些瀏覽日誌，存資料庫浪費，直接存硬碟又怕到時候操作效率低。 此時，

分散式釋出訂閱訊息系統—Apache Kafka

1.什麼是Kafka Kafka是一種高吞吐量的分散式釋出訂閱訊息系統，它可以處理消費者規模的網站中的所有動作流資料。 這種動作（網頁瀏覽，搜尋和其他使用者的行動）是在現代網路上的許多社會功能的一個關鍵因素。 這些資料通常是由於吞吐量的要求而通過處理日誌和日誌聚合來解決。

RocketMQ中介軟體訊息佇列在Maven專案中的配置使用操作 (分散式釋出訂閱訊息系統)

一、專案引用 <dependency>     <groupId>com.foriseland.fjf.mq</groupId>     <artifactI

釋出訂閱訊息系統--kafka的解析。

Kafka介紹： Kafka是由Apache軟體基金會開發的一個開源流處理平臺，由Scala和Java編寫。Kafka是一種高吞吐量的分散式釋出訂閱訊息系統，它可以處理消費者規模的網站中的所有動作流資料。這種動作（網頁瀏覽，搜尋和其他使用者的行動）是在現代網路上的許多社會功

Apache Pulsar：雅虎開發的企業級釋出訂閱訊息系統

英文原文：https://streaml.io/blog/intro-to-pulsar/中文翻

釋出 訂閱 訊息系統

1.從監聽與釋出說起 我們寫js程式碼的時候都知道有這樣的事件：我們註冊一個click方法 ，此時我們就為這個按鈕添加了“監聽”，基於“點選”事件的監聽。以此來實現點選按鈕提交表單資料的目的，在這裡，

分散式訊息系統Kafka初步

終於可以寫kafka的文章了，Mina的相關文章我已經做了索引，在我的部落格中置頂了，大家可以方便的找到。從這一篇開始分散式訊息系統的入門。 在我們大量使用分散式資料庫、分散式計算叢集的時候，是否會遇到這樣的一些問題： l  我想分析一下使用者行為（pageviews

高併發面試必問：分散式訊息系統Kafka簡介

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

分散式訊息系統Kafka簡介

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

分散式訊息系統Kafka介紹

轉載自董的部落格，文章地址：http://dongxicheng.org/search-engine/kafka/ 1、  概述 Kafka是Linkedin於2010年12月份開源的訊息系統，它主要用於處理活躍的流式資料。活躍的流式資料在web網站應用中非常常見，這些

ActiveMQ實戰之 Topic釋出訂閱訊息

分享一下我老師大神的人工智慧教程！零基礎，通俗易懂！http://blog.csdn.net/jiangjunshow 也歡迎大家轉載本篇文章。分享知識，造福人民，實現我們中華民族偉大復興！        

ActiveMQ之釋出- 訂閱訊息模式實現

一、概念 釋出者/訂閱者模型支援向一個特定的訊息主題釋出訊息。0或多個訂閱者可能對接收來自特定訊息主題的訊息感興趣。在這種模型下，釋出者和訂閱者彼此不知道對方。這種模式好比是匿名公告板。這種模式被概括為：多個消費者可以獲得訊息 在釋出者和訂閱者之間存在時間依賴性。釋出者需要建

Akka-Cluster（2）- distributed pub/sub mechanism 分散式釋出/訂閱機制

   上期我們介紹了cluster singleton，它的作用是保證在一個叢集環境裡永遠會有唯一一個singleton例項存在。具體使用方式是在叢集所有節點部署ClusterSingletonManager，由叢集中的leader節點選定其中一個節點並指示上面的Cluste

[Python]zeromq：socket request/receive, 釋出/訂閱訊息, 程序間通訊

Request/Response Server: import zmq context = zmq.Context() socket = context.socket(zmq.REP) socket.bind("tcp://*:5555") while True: mes