2019獨(dú)角獸企業(yè)重金招聘Python工程師標(biāo)準(zhǔn)>>>

基本概念

kafka中的重要角色
??broker：一臺(tái)kafka服務(wù)器就是一個(gè)broker，一個(gè)集群可有多個(gè)broker，一個(gè)broker可以容納多個(gè)topic
??topic：可以理解為一個(gè)消息隊(duì)列的名字
??partition：分區(qū)，為了實(shí)現(xiàn)擴(kuò)展性，一個(gè)topic可以分布到多個(gè)broker上，一個(gè)topic可以被分成多個(gè)partition，partition中的每條消息 都會(huì)被分配一個(gè)有序的id（offset）。kafka只保證按一個(gè)partition中的順序?qū)⑾l(fā)給consumer，不保證一個(gè)topic的整體的順序。也就是說(shuō)，一個(gè)topic在集群中可以有多個(gè)partition 。kafka有Key Hash算法和Round Robin算法兩種分區(qū)策略。
??producer：消息的生產(chǎn)者，是向kafka發(fā)消息的客戶端
??consumer：消息消費(fèi)者，向broker取消息的客戶端
??offset：偏移量，用來(lái)記錄consumer消費(fèi)消息的位置
??Consumer Group：消費(fèi)組，消息系統(tǒng)有兩類，一是廣播，二是訂閱發(fā)布。

編碼實(shí)現(xiàn)

??創(chuàng)建一個(gè)生產(chǎn)者

package sancen.kafkaimport java.util.Propertiesimport kafka.producer.{KeyedMessage, Producer, ProducerConfig}/*** 類名 ProducerDemo* 作者 彭三青* 創(chuàng)建時(shí)間 2018-11-26 9:49* 版本 1.0* 描述: $ 實(shí)現(xiàn)一個(gè)生產(chǎn)者，把模擬數(shù)據(jù)發(fā)送到kafka集群*/object ProducerDemo {def main(args: Array[String]): Unit = {// 定義一個(gè)接收數(shù)據(jù)的topicval topic = "test"// 創(chuàng)建配置信息val props = new Properties()// 指定序列化類props.put("serializer.class", "kafka.serializer.StringEncoder")// 指定kafka列表props.put("metadata.broker.list", "SC01:9092, SC01:9092, SC03:9092")// 設(shè)置發(fā)送數(shù)據(jù)后的響應(yīng)方式props.put("request.required.acks", "0")// 指定分區(qū)器// props.put("partitioner.class", "kafka.producer.DefaultPartitioner// 自定義分區(qū)器props.put("partitioner.class", "day01.kafka.CustomPartitioner")// 創(chuàng)建producer對(duì)象val config: ProducerConfig = new ProducerConfig(props)// 創(chuàng)建生產(chǎn)者對(duì)象val producer: Producer[String, String] = new Producer(config)// 模擬數(shù)據(jù)for(i <- 1 to 10000){val msg = s"$i : producer send data"producer.send(new KeyedMessage[String, String](topic, msg)) //key偏移量，也可以給空 v實(shí)際的數(shù)據(jù)Thread.sleep(500)}} }

??創(chuàng)建消費(fèi)者

package sancen.kafkaimport java.util.Properties import java.util.concurrent.{ExecutorService, Executors}import kafka.consumer._import scala.collection.mutable/*** 類名 ConsumerDemo* 作者 彭三青* 創(chuàng)建時(shí)間 2018-11-26 10:08* 版本 1.0* 描述: $ 創(chuàng)建一個(gè)Consumer消費(fèi)kafka的數(shù)據(jù)*/class ConsumerDemo(val consumer: String, val stream: KafkaStream[Array[Byte], Array[Byte]]) extends Runnable{override def run(): Unit = {val it: ConsumerIterator[Array[Byte], Array[Byte]] = stream.iterator()while (it.hasNext()){val data = it.next()val topic = data.topicval partition = data.partitionval offset = data.offsetval msg: String = new String(data.message())println(s"Consumer:$consumer, topic:$topic, partiton:$partition, offset:$offset, msg:$msg")}} }object ConsumerDemo {def main(args: Array[String]): Unit = {// 定義獲取的topicval topic = "test"// 定義一個(gè)map，用來(lái)存儲(chǔ)多個(gè)topic key:topic名稱，value:指定線程數(shù)用來(lái)獲取topic的數(shù)據(jù)val topics = new mutable.HashMap[String, Int]() // 要求就要傳一個(gè)map，可以放一個(gè)或者多個(gè)topictopics.put(topic, 2)// 配置信息val props = new Properties()// 指定consumer組名props.put("group.id", "group02")// 指定zk列表props.put("zookeeper.connect", "SC01:2181,SC02:2181,SC03:2181")// 指定offset異常時(shí)需要獲取的offset值props.put("auto.offset.reset", "smallest")// 創(chuàng)建配置信息val config = new ConsumerConfig(props)// 創(chuàng)建consumer對(duì)象val consumer: ConsumerConnector = Consumer.create(config)// 獲取數(shù)據(jù),返回的map類型中key：topic名稱，value：topic對(duì)應(yīng)的數(shù)據(jù)val streams: collection.Map[String, List[KafkaStream[Array[Byte], Array[Byte]]]] = consumer.createMessageStreams(topics)// 獲取指定topic的數(shù)據(jù)val stream: Option[List[KafkaStream[Array[Byte], Array[Byte]]]] = streams.get(topic)// 創(chuàng)建固定大小的線程池val pool: ExecutorService = Executors.newFixedThreadPool(3)for(i <- 0 until stream.size){pool.execute(new ConsumerDemo(s"Consumer:$i", stream.get(i)))}} }

??創(chuàng)建自定義分區(qū)類

package sancen.kafkaimport kafka.producer.Partitioner import kafka.utils.VerifiableProperties import org.apache.kafka.common.utils.Utils/*** 類名 CustomPartitioner* 作者 彭三青* 創(chuàng)建時(shí)間 2018-11-26 20:29* 版本 1.0* 描述: $*/// 要實(shí)現(xiàn)自定義分區(qū)器必須要繼承Partitioner class CustomPartitioner(props: VerifiableProperties) extends Partitioner{override def partition(key: Any, numPartitions: Int): Int = {Utils.abs(key.hashCode) % numPartitions} }

程序測(cè)試

??后臺(tái)啟動(dòng)kafka集群

kafka-server-start.sh kafka_2.11-0.9.0.1/config/server.properties &

??在kafka集群上創(chuàng)建一個(gè)名為test的topic，指定分區(qū)為2，一般一個(gè)topic對(duì)應(yīng)一個(gè)分區(qū)

kafka-topics.sh --create --zookeeper SC01:2181 --replication-factor 2 --partitions 2 --topic test

??分別運(yùn)行ProducerDemo和ConsumerDemo則可以在ConsumerDemo端窗口打印出信息

轉(zhuǎn)載于:https://my.oschina.net/u/3875806/blog/2962794

總結(jié)

以上是生活随笔為你收集整理的kafka模拟生产-消费者以及自定义分区的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。