使用java實(shí)現(xiàn)Kafka的生產(chǎn)者

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869

package com.lisg.kafkatest;import java.util.Properties;import kafka.javaapi.producer.Producer;import kafka.producer.KeyedMessage;import kafka.producer.Partitioner;import kafka.producer.ProducerConfig;import kafka.serializer.StringEncoder;/**?* Kafka生產(chǎn)者?* @author lisg?*?*/public class KafkaProducer {????public static void main(String[] args) {?????????????????Properties props = new Properties();????????//根據(jù)這個(gè)配置獲取metadata,不必是kafka集群上的所有broker,但最好至少有兩個(gè)????????props.put("metadata.broker.list", "vm1:9092,vm2:9092");????????//消息傳遞到broker時(shí)的序列化方式????????props.put("serializer.class", StringEncoder.class.getName());????????//zk集群????????props.put("zookeeper.connect", "vm1:2181");????????//是否獲取反饋????????//0是不獲取反饋(消息有可能傳輸失敗)????????//1是獲取消息傳遞給leader后反饋(其他副本有可能接受消息失敗)????????//-1是所有in-sync replicas接受到消息時(shí)的反饋????????props.put("request.required.acks", "1");//????? props.put("partitioner.class", MyPartition.class.getName());?????????????????//創(chuàng)建Kafka的生產(chǎn)者, key是消息的key的類型, value是消息的類型????????Producer<Integer, String> producer = new Producer<Integer, String>(????????????????new ProducerConfig(props));?????????????????int count = 0;????????while(true) {????????????String message = "message-" + ++count;????????????//消息主題是test????????????KeyedMessage<Integer, String> keyedMessage = new KeyedMessage<Integer, String>("test", message);????????????//message可以帶key, 根據(jù)key來將消息分配到指定區(qū), 如果沒有key則隨機(jī)分配到某個(gè)區(qū)//????????? KeyedMessage<Integer, String> keyedMessage = new KeyedMessage<Integer, String>("test", 1, message);????????????producer.send(keyedMessage);????????????System.out.println("send: " + message);????????????try {????????????????Thread.sleep(1000);????????????} catch (InterruptedException e) {????????????????e.printStackTrace();????????????}????????}?????????//????? producer.close();????} }/**?* 自定義分區(qū)類?*?*/class MyPartition implements Partitioner {????public int partition(Object key, int numPartitions) {????????return key.hashCode()%numPartitions;????}?????}

來自為知筆記(Wiz)

附件列表

?

轉(zhuǎn)載于:https://www.cnblogs.com/lishouguang/p/4560559.html

總結(jié)

以上是生活随笔為你收集整理的java实现Kafka生产者示例的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

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