Kafka使用Java進行Producer和Consumer程式設計

阿新 • • 發佈：2019-01-12

比較舊的kafka_2.10-0.8.2.0版本：（參考自http://chengjianxiaoxue.iteye.com/blog/2190488）

生產者程式碼：

import java.util.Properties;  
import java.util.concurrent.TimeUnit;  
  
import kafka.javaapi.producer.Producer;  
import kafka.producer.KeyedMessage;  
import kafka.producer.ProducerConfig;
import kafka.serializer.StringEncoder;  

public class kafkaProducer extends Thread{  
  
    private String topic;  
      
    public kafkaProducer(String topic){  
        super();  
        this.topic = topic;  
    }  
      
    @Override
    public void run() {  
        Producer producer = createProducer();  
        int i=0;
        while(true){
            producer.send(new KeyedMessage<Integer, String>(topic, "message: " + i++));  
            try {
                TimeUnit.SECONDS.sleep(1);  
            } catch (InterruptedException e) {  
                e.printStackTrace();  
            }  
        }
    }  
 
    private Producer createProducer() {  
        Properties properties = new Properties();  
        properties.put("zookeeper.connect", "192.168.205.153:2181");//宣告zk  
        properties.put("serializer.class", StringEncoder.class.getName());  
        properties.put("metadata.broker.list", "192.168.205.153:9092");// 宣告kafka broker  
        return new Producer<Integer, String>(new ProducerConfig(properties));  
     }  
      
      
    public static void main(String[] args) {  
        new kafkaProducer("test").start();// 使用kafka叢集中建立好的主題 test   
    }  

}

消費者程式碼：

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

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

/** 
 * 接收資料 
接收到: message: 10 
接收到: message: 11 
接收到: message: 12 
接收到: message: 13 
接收到: message: 14 
 */  
public class kafkaConsumer extends Thread{  
  
    private String topic;  
      
    public kafkaConsumer(String topic){  
        super();  
        this.topic = topic;  
    }  
      
    @Override  
    public void run() {  
        ConsumerConnector consumer = createConsumer();  
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();  
        topicCountMap.put(topic, 1); // 1表示consumer thread執行緒數量
         Map<String, List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);  
         KafkaStream<byte[], byte[]> stream = messageStreams.get(topic).get(0);// 獲取每次接收到的這個資料  
         ConsumerIterator<byte[], byte[]> iterator =  stream.iterator();  
         while(iterator.hasNext()){  
             String message = new String(iterator.next().message());  
             System.out.println("接收到: " + message);  
         }  
    }  
  
    private ConsumerConnector createConsumer() {  
        Properties properties = new Properties();  
        properties.put("zookeeper.connect", "192.168.205.153:2181");//宣告zk  
        properties.put("group.id", "group1");// 必須要使用別的組名稱， 如果生產者和消費者都在同一組，則不能訪問同一組內的topic資料  
        return Consumer.createJavaConsumerConnector(new ConsumerConfig(properties));  
     }  
      
    public static void main(String[] args) {  
        new kafkaConsumer("test").start();// 使用kafka叢集中建立好的主題 test   
    }  
       
}

注意：以上程式碼在kafka_2.10-0.8.2.0和kafka_2.11-0.9.0.1中都能執行，因為高版本會相容低版本，所以高版本的程式碼不一定能在低版本里執行，而低版本的程式碼一般都能在高版本中執行。

kafka_2.11-0.9.0.1版本：

例子一：（程式碼參考自http://blog.csdn.net/lnho2015/article/details/51353936）

生產者程式碼：

import org.apache.kafka.clients.producer.KafkaProducer;  
import org.apache.kafka.clients.producer.Producer;  
import org.apache.kafka.clients.producer.ProducerRecord;  
  
import java.util.Properties;  
  
public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();  
        props.put("bootstrap.servers", "h153:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);
        for(int i = 0; i < 100; i++)
//            producer.send(new ProducerRecord<>("test", Integer.toString(i), Integer.toString(i)));
        	producer.send(new ProducerRecord<String, String>("test", "Hello"));
        	producer.close();
    }
}

消費者程式碼：

import org.apache.kafka.clients.consumer.ConsumerRecord;  
import org.apache.kafka.clients.consumer.ConsumerRecords;  
import org.apache.kafka.clients.consumer.KafkaConsumer;  
  
import java.util.Arrays;  
import java.util.Properties;  
  
public class KafkaConsumerExample {  
    public static void main(String[] args) {  
        Properties props = new Properties();  
        props.put("bootstrap.servers", "h153:9092");
        props.put("group.id", "test");  
        props.put("enable.auto.commit", "true");  
        props.put("auto.commit.interval.ms", "1000");  
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("test"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);  
            for (ConsumerRecord<String, String> record : records)  
                System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value());  
        }
    }
}

注意1：一開始用myeclipse匯入kafka_2.10-0.8.2.0jar包正常，在匯入kafka_2.11-0.9.0.1後雖然報錯但能生產資料，在匯入kafka_2.10-0.10.1.0和kafka_2.11-0.11.0jar包就直接報錯了而不產生資料。在匯入kafka_2.11-0.9.0.1jar包報這個錯：

[2017-10-30 19:08:39,779] ERROR Uncaught error in kafka producer I/O thread: (org.apache.kafka.clients.producer.internals.Sender:143)
org.apache.kafka.common.protocol.types.SchemaException: Error reading field 'throttle_time_ms': java.nio.BufferUnderflowException
at org.apache.kafka.common.protocol.types.Schema.read(Schema.java:71)
at org.apache.kafka.clients.NetworkClient.handleCompletedReceives(NetworkClient.java:439)
at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:265)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:216)
at org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:141)
at java.lang.Thread.run(Thread.java:724)

原因：我在myeclipse中匯入的是kafka_2.11-0.9.0.1的jar包，而在Linux虛擬機器中安裝的卻是kafka_2.10-0.8.2.0

注意2：當我在Linux本地上執行生產者程式碼卻報這個錯

[[email protected] q1]$ javac KafkaProducerExample.java 
KafkaProducerExample.java:1: error: package org.apache.kafka.clients.producer does not exist
import org.apache.kafka.clients.producer.KafkaProducer;    
                                        ^
KafkaProducerExample.java:2: error: package org.apache.kafka.clients.producer does not exist
import org.apache.kafka.clients.producer.Producer;    
                                        ^
KafkaProducerExample.java:3: error: package org.apache.kafka.clients.producer does not exist
import org.apache.kafka.clients.producer.ProducerRecord;    
                                        ^
KafkaProducerExample.java:19: error: cannot find symbol
        Producer<String, String> producer = new KafkaProducer<>(props);  
        ^
  symbol:   class Producer
  location: class KafkaProducerExample
KafkaProducerExample.java:19: error: cannot find symbol
        Producer<String, String> producer = new KafkaProducer<>(props);  
                                                ^
  symbol:   class KafkaProducer
  location: class KafkaProducerExample
KafkaProducerExample.java:21: error: cannot find symbol
            producer.send(new ProducerRecord<>("test", Integer.toString(i), Integer.toString(i)));
                              ^
  symbol:   class ProducerRecord
  location: class KafkaProducerExample
6 errors

解決：我當時裝的Java版本是jdk1.8.0_91，換為jdk1.7.0_25，並且在~/.bash_profile中新增如Kafkajar包到環境變數中，如下：

export CLASSPATH=.:/home/hadoop/hbase-1.0.0/lib/*:/home/hadoop/kafka_2.11-0.9.0.1/libs/*

例子二：（參考自http://blog.csdn.net/zero__007/article/details/51068165）

生產者程式碼：

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;  
import org.apache.kafka.clients.producer.Producer;  
import org.apache.kafka.clients.producer.ProducerRecord;  
import org.apache.kafka.clients.producer.RecordMetadata;
  
import java.util.Properties;  

public class Produce {
    public static void main(String[] args) {
    	String topic = "test";
    	Properties props = new Properties();  
    	props.put("bootstrap.servers", "h153:9092");  
    	props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
    	props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");  
    	props.put("acks", "all");  
    	props.put("retries", 1);  
    	 
    	Producer<String, String> producer = new KafkaProducer<String, String>(props);  
    	producer.send(new ProducerRecord<String, String>(topic, "Hello"));
    	  
    	producer.send(new ProducerRecord<String, String>(topic, "World"), new Callback() {  
    	    @Override  
    	    public void onCompletion(RecordMetadata metadata, Exception e) {  
    	        if (e != null) {
    	            e.printStackTrace();
    	        } else {
    	            System.out.println(metadata.toString());//[email protected]  
    	            System.out.println(metadata.offset());//1  
    	        }
    	    }
    	});  
    	producer.flush();
    	producer.close();
    }
}

消費者程式碼：

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.TopicPartition;
  
import java.util.Arrays;  
import java.util.Properties;  
  
public class Consumer {  
    public static void main(String[] args) {  
    	String topic = "test";
    	Properties props = new Properties();  
    	props.put("bootstrap.servers", " h153:9092");  
    	props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
    	props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");  
    	props.setProperty("group.id", "0");  
    	props.setProperty("enable.auto.commit", "true");  
    	props.setProperty("auto.offset.reset", "earliest");  
    	  
    	KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);  
    	consumer.subscribe(Arrays.asList(topic));  
    	  
    	for (int i = 0; i < 2; i++) {  
    	    ConsumerRecords<String, String> records = consumer.poll(1000);  
    	    System.out.println(records.count());  
    	    for (ConsumerRecord<String, String> record : records) {  
    	        System.out.println(record);  
//    	        consumer.seekToBeginning(new TopicPartition(record.topic(), record.partition()));
    	    }  
    	}
    }
}

注意：auto.offset.reset這裡設定為earliest，是為了consumer能夠從頭開始讀取內容即offset=0開始,在org.apache.kafka.clients.consumer.ConsumerConfig中對其意義的描述如下：What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server (e.g. because that data has been deleted): earliest: automatically reset the offset to the earliest offset；latest: automatically reset the offset to the latest offset；none: throw exception to the consumer if no previous offset is found for the consumer's group；anything else: throw exception to the consumer。consumer.seekToBeginning也可以設定offset，但是跟原始碼發現，This function evaluates lazily, seeking to the final offset in all partitions only when {@link #poll(long)} or {@link #position(TopicPartition)} are called.也就是說seekToBeginning()的設定要生效的話，必須在poll或則position方法呼叫後設置seekToBeginning()才行。

補充（可參考這篇文章：http://blog.csdn.net/lishuangzhe7047/article/details/74530417）：

earliest
當各分割槽下有已提交的offset時，從提交的offset開始消費；無提交的offset時，從頭開始消費
latest
當各分割槽下有已提交的offset時，從提交的offset開始消費；無提交的offset時，消費新產生的該分割槽下的資料
none
topic各分割槽都存在已提交的offset時，從offset後開始消費；只要有一個分割槽不存在已提交的offset，則丟擲異常

Kafka使用Java進行Producer和Consumer程式設計

Kafka使用Java進行Producer和Consumer程式設計

Kafka學習整理七(producer和consumer程式設計實踐)

Java連接ActiveMQ代碼示例(Producer和Consumer)

kafka執行Producer和Consumer時出現Failed to load class org.slf4j.impl.StaticLoggerBinder錯誤

安裝部署（六） Kafka叢集安裝部署以及Producer和Consumer的JAVA程式碼測試

Kafka入門，producer和consumer與hive

RocketMQ中Producer和Consumer啟動出錯

kafka客戶端Producer和Consumer關於自定義訊息序列和反序列

使用Scratch2和ROS進行機器人圖形化程式設計學習

使用 C++ 和 MFC 進行多執行緒程式設計

Kafka系列3-python版本producer生產者和consumer消費者例項

rabbit - producer的confirm和consumer的ack模式

BOS項目第7天(shiro權限框架進行認證和授權)

什麽是私有密鑰密碼技術——密鑰加密算法采用同一把密鑰進行加密和解密

如何確定Kafka的分區數、key和consumer線程數

C/C++使用openssl進行摘要和加密解密（md5, sha256, des, rsa）

在windows 下使用eclipse進行編譯和燒寫

蜜罐技術——通過布置一些作為誘餌的主機、網絡服務或者信息，誘使攻擊方對它們實施攻擊，從而可以對攻擊行為進行捕獲和分析

網絡安全熱門話題——如何對被（已經/正在）入侵網站進行檢測和防範

Kafka 學習筆記之 Kafka0.11之console-producer/console-consumer

Kafka使用Java進行Producer和Consumer程式設計

相關推薦