Apache Kafka - Ví dụ về nhà sản xuất đơn giản

Hãy để chúng tôi tạo một ứng dụng để xuất bản và sử dụng thư bằng máy khách Java. Ứng dụng khách nhà sản xuất Kafka bao gồm các API sau.

API của nhà sản xuất Kafka

Hãy để chúng tôi hiểu bộ API nhà sản xuất Kafka quan trọng nhất trong phần này. Phần trung tâm của API KafkaProduction là lớp KafkaProductioner . Lớp KafkaProductioner cung cấp một tùy chọn để kết nối một nhà môi giới Kafka trong phương thức khởi tạo của nó với các phương thức sau.

• Lớp KafkaProductioner cung cấp phương thức gửi để gửi các thông điệp không đồng bộ đến một chủ đề. Chữ ký của send () như sau

producer.send(new ProducerRecord<byte[],byte[]>(topic, 
partition, key1, value1) , callback);

• ProducerRecord - Người sản xuất quản lý một bộ đệm các bản ghi đang chờ được gửi đi.

• Callback - Lệnh gọi lại do người dùng cung cấp để thực thi khi bản ghi đã được máy chủ xác nhận (null cho biết không có lệnh gọi lại).

• Lớp KafkaProductioner cung cấp một phương thức tuôn ra để đảm bảo rằng tất cả các thư đã gửi trước đó đã thực sự được hoàn thành. Cú pháp của phương thức flush như sau:

public void flush()

• Lớp KafkaProductioner cung cấp phương thức partitionFor, giúp lấy siêu dữ liệu phân vùng cho một chủ đề nhất định. Điều này có thể được sử dụng để phân vùng tùy chỉnh. Chữ ký của phương pháp này như sau:

public Map metrics()

Nó trả về bản đồ các chỉ số nội bộ do nhà sản xuất duy trì.

• public void close () - Lớp KafkaProductioner cung cấp các khối phương thức đóng cho đến khi hoàn thành tất cả các yêu cầu đã gửi trước đó.

API nhà sản xuất

Phần trung tâm của API Producer là lớp Producer . Lớp Producer cung cấp một tùy chọn để kết nối nhà môi giới Kafka trong phương thức khởi tạo của nó bằng các phương thức sau.

Lớp nhà sản xuất

Lớp nhà sản xuất cung cấp phương thức gửi tới send tin nhắn cho một hoặc nhiều chủ đề bằng cách sử dụng các chữ ký sau.

public void send(KeyedMessaget<k,v> message) 
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(List<KeyedMessage<k,v>>messages)
- sends data to multiple topics.
Properties prop = new Properties();
prop.put(producer.type,”async”)
ProducerConfig config = new ProducerConfig(prop);

Có hai loại nhà sản xuất - Sync và Async.

Cấu hình API tương tự cũng áp dụng cho nhà sản xuất Đồng bộ hóa . Sự khác biệt giữa chúng là một nhà sản xuất đồng bộ gửi tin nhắn trực tiếp, nhưng gửi tin nhắn trong nền. Trình sản xuất không đồng bộ được ưu tiên khi bạn muốn thông lượng cao hơn. Trong các phiên bản trước như 0.8, trình sản xuất không đồng bộ không có lệnh gọi lại cho send () để đăng ký trình xử lý lỗi. Điều này chỉ có sẵn trong phiên bản 0.9 hiện tại.

public void close ()

Lớp nhà sản xuất cung cấp close phương pháp đóng kết nối nhóm nhà sản xuất với tất cả các nhà sản xuất Kafka.

Thiết lập cấu hình

Cài đặt cấu hình chính của API Producer được liệt kê trong bảng sau để có tình trạng tốt hơn -

S. không	Cài đặt cấu hình và mô tả
1	client.id xác định ứng dụng của nhà sản xuất
2	producer.type đồng bộ hoặc không đồng bộ
3	acks Cấu hình acks kiểm soát các tiêu chí theo yêu cầu của nhà sản xuất đã hoàn tất.
4	retries Nếu yêu cầu của nhà sản xuất không thành công, thì tự động thử lại với giá trị cụ thể.
5	bootstrap.servers bootstrapping danh sách các nhà môi giới.
6	linger.ms nếu bạn muốn giảm số lượng yêu cầu, bạn có thể đặt linger.ms thành giá trị lớn hơn một số giá trị.
7	key.serializer Phím cho giao diện bộ nối tiếp.
số 8	value.serializer giá trị cho giao diện bộ nối tiếp.
9	batch.size Kích thước đệm.
10	buffer.memory kiểm soát tổng dung lượng bộ nhớ có sẵn cho nhà sản xuất để buff-ering.

API ProducerRecord

ProducerRecord là một cặp khóa / giá trị được gửi đến phương thức khởi tạo lớp Kafka cluster.ProductionerRecord để tạo bản ghi với các cặp phân vùng, khóa và giá trị bằng cách sử dụng chữ ký sau.

public ProducerRecord (string topic, int partition, k key, v value)

• Topic - tên chủ đề do người dùng xác định sẽ được thêm vào để ghi.

• Partition - số lượng phân vùng

• Key - Chìa khóa sẽ có trong hồ sơ.

• Value - Ghi lại nội dung

public ProducerRecord (string topic, k key, v value)

Phương thức khởi tạo lớp ProducerRecord được sử dụng để tạo một bản ghi với các cặp khóa, giá trị và không có phân vùng.

• Topic - Tạo chủ đề để gán bản ghi.

• Key - chìa khóa cho bản ghi.

• Value - ghi lại nội dung.

public ProducerRecord (string topic, v value)

Lớp ProducerRecord tạo một bản ghi không có phân vùng và khóa.

• Topic - tạo chủ đề.

• Value - ghi lại nội dung.

Các phương thức của lớp ProducerRecord được liệt kê trong bảng sau:

S. không	Phương thức và mô tả lớp học
1	public string topic() Chủ đề sẽ thêm vào hồ sơ.
2	public K key() Chìa khóa sẽ được bao gồm trong hồ sơ. Nếu không có khóa như vậy, null sẽ được chuyển lại ở đây.
3	public V value() Ghi lại nội dung.
4	partition() Số lượng phân vùng cho bản ghi

Ứng dụng đơn giản

Trước khi tạo ứng dụng, trước tiên hãy khởi động ZooKeeper và Kafka broker sau đó tạo chủ đề của riêng bạn trong Kafka broker bằng lệnh create topic. Sau đó, tạo một lớp java có tên Sim-pleProductioner.java và nhập mã sau.

//import util.properties packages
import java.util.Properties;

//import simple producer packages
import org.apache.kafka.clients.producer.Producer;

//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;

//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;

//Create java class named “SimpleProducer”
public class SimpleProducer {
   
   public static void main(String[] args) throws Exception{
      
      // Check arguments length value
      if(args.length == 0){
         System.out.println("Enter topic name”);
         return;
      }
      
      //Assign topicName to string variable
      String topicName = args[0].toString();
      
      // create instance for properties to access producer configs   
      Properties props = new Properties();
      
      //Assign localhost id
      props.put("bootstrap.servers", “localhost:9092");
      
      //Set acknowledgements for producer requests.      
      props.put("acks", “all");
      
      //If the request fails, the producer can automatically retry,
      props.put("retries", 0);
      
      //Specify buffer size in config
      props.put("batch.size", 16384);
      
      //Reduce the no of requests less than 0   
      props.put("linger.ms", 1);
      
      //The buffer.memory controls the total amount of memory available to the producer for buffering.   
      props.put("buffer.memory", 33554432);
      
      props.put("key.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
         
      props.put("value.serializer", 
         "org.apache.kafka.common.serializa-tion.StringSerializer");
      
      Producer<String, String> producer = new KafkaProducer
         <String, String>(props);
            
      for(int i = 0; i < 10; i++)
         producer.send(new ProducerRecord<String, String>(topicName, 
            Integer.toString(i), Integer.toString(i)));
               System.out.println(“Message sent successfully”);
               producer.close();
   }
}

Compilation - Ứng dụng có thể được biên dịch bằng lệnh sau.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution - Ứng dụng có thể được thực thi bằng lệnh sau.

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleProducer <topic-name>

Output

Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.
>> bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic <topic-name> —from-beginning
1
2
3
4
5
6
7
8
9
10

Ví dụ về người tiêu dùng đơn giản

Hiện tại, chúng tôi đã tạo một trình sản xuất để gửi tin nhắn đến Kafka cluster. Bây giờ chúng ta hãy tạo một người tiêu dùng để sử dụng các thông điệp tạo thành cụm Kafka. API KafkaConsumer được sử dụng để sử dụng các thông báo từ cụm Kafka. Phương thức khởi tạo lớp KafkaConsumer được định nghĩa bên dưới.

public KafkaConsumer(java.util.Map<java.lang.String,java.lang.Object> configs)

configs - Trả lại bản đồ cấu hình người tiêu dùng.

Lớp KafkaConsumer có các phương thức quan trọng sau được liệt kê trong bảng dưới đây.

S. không	Phương pháp và Mô tả
1	public java.util.Set<TopicPar-tition> assignment() Nhận tập hợp các phân vùng hiện được chỉ định bởi con-sumer.
2	public string subscription() Đăng ký danh sách các chủ đề nhất định để nhận các phân vùng động như đã ký.
3	public void sub-scribe(java.util.List<java.lang.String> topics, ConsumerRe-balanceListener listener) Đăng ký danh sách các chủ đề nhất định để nhận các phân vùng động như đã ký.
4	public void unsubscribe() Hủy đăng ký các chủ đề khỏi danh sách phân vùng đã cho.
5	public void sub-scribe(java.util.List<java.lang.String> topics) Đăng ký danh sách các chủ đề nhất định để nhận các phân vùng động như đã ký. Nếu danh sách các chủ đề nhất định trống, nó được coi giống như hủy đăng ký ().
6	public void sub-scribe(java.util.regex.Pattern pattern, ConsumerRebalanceLis-tener listener) Mẫu đối số đề cập đến mẫu đăng ký ở định dạng biểu thức chính quy và đối số trình nghe nhận thông báo từ mẫu đăng ký.
7	public void as-sign(java.util.List<TopicParti-tion> partitions) Chỉ định thủ công danh sách các phân vùng cho khách hàng.
số 8	poll() Tìm nạp dữ liệu cho các chủ đề hoặc phân vùng được chỉ định bằng cách sử dụng một trong các API đăng ký / gán. Điều này sẽ trả về lỗi, nếu các chủ đề không được đăng ký trước khi bỏ phiếu cho dữ liệu.
9	public void commitSync() Mức bù cam kết được trả lại trong cuộc thăm dò cuối cùng () cho tất cả danh sách chủ đề và phân vùng được ghi chép phụ. Thao tác tương tự cũng được áp dụng cho commitAsyn ().
10	public void seek(TopicPartition partition, long offset) Tìm nạp giá trị bù đắp hiện tại mà người tiêu dùng sẽ sử dụng trong phương thức thăm dò ý kiến ​​() tiếp theo.
11	public void resume() Tiếp tục các phân vùng bị tạm dừng.
12	public void wakeup() Đánh thức người tiêu dùng.

API ConsumerRecord

API ConsumerRecord được sử dụng để nhận các bản ghi từ cụm Kafka. API này bao gồm tên chủ đề, số phân vùng mà bản ghi đang được nhận và phần bù trỏ đến bản ghi trong phân vùng Kafka. Lớp ConsumerRecord được sử dụng để tạo một bản ghi người tiêu dùng với tên chủ đề cụ thể, số lượng phân vùng và các cặp <key, value>. Nó có chữ ký sau.

public ConsumerRecord(string topic,int partition, long offset,K key, V value)

• Topic - Tên chủ đề cho hồ sơ người tiêu dùng nhận được từ cụm Kafka.

• Partition - Phân vùng cho chủ đề.

• Key - Khoá của bản ghi, nếu không có khoá nào tồn tại null sẽ được trả về.

• Value - Ghi lại nội dung.

API ConsumerRecords

API ConsumerRecords hoạt động như một vùng chứa cho ConsumerRecord. API này được sử dụng để giữ danh sách ConsumerRecord trên mỗi phân vùng cho một chủ đề cụ thể. Constructor của nó được định nghĩa bên dưới.

public ConsumerRecords(java.util.Map<TopicPartition,java.util.List
<Consumer-Record>K,V>>> records)

• TopicPartition - Trả lại bản đồ phân vùng cho một chủ đề cụ thể.

• Records - Danh sách trả lại của ConsumerRecord.

Lớp ConsumerRecords có các phương thức sau được định nghĩa.

S. không	Phương pháp và Mô tả
1	public int count() Số lượng bản ghi cho tất cả các chủ đề.
2	public Set partitions() Tập hợp các phân vùng có dữ liệu trong tập bản ghi này (nếu không có dữ liệu nào được trả về thì tập đó trống).
3	public Iterator iterator() Iterator cho phép bạn xoay vòng qua một tập hợp, thu thập hoặc di chuyển lại các phần tử.
4	public List records() Nhận danh sách các bản ghi cho phân vùng đã cho.

Thiết lập cấu hình

Cài đặt cấu hình cho cài đặt cấu hình chính API khách hàng tiêu dùng được liệt kê bên dưới -

S. không	Cài đặt và Mô tả
1	bootstrap.servers Bootstrapping danh sách các nhà môi giới.
2	group.id Chỉ định một người tiêu dùng cá nhân cho một nhóm.
3	enable.auto.commit Bật cam kết tự động cho các giá trị bù nếu giá trị là true, nếu không thì không được cam kết.
4	auto.commit.interval.ms Trả về tần suất các hiệu số đã tiêu thụ được cập nhật được ghi vào ZooKeeper.
5	session.timeout.ms Cho biết Kafka sẽ đợi bao nhiêu mili giây để ZooKeeper phản hồi một yêu cầu (đọc hoặc ghi) trước khi từ bỏ và tiếp tục sử dụng tin nhắn.

Ứng dụng SimpleConsumer

Các bước ứng dụng của nhà sản xuất vẫn giữ nguyên ở đây. Đầu tiên, hãy khởi động nhà môi giới ZooKeeper và Kafka của bạn. Sau đó, tạo một ứng dụng SimpleConsumer với lớp java có tên SimpleCon-sumer.java và nhập đoạn mã sau.

import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class SimpleConsumer {
   public static void main(String[] args) throws Exception {
      if(args.length == 0){
         System.out.println("Enter topic name");
         return;
      }
      //Kafka consumer configuration settings
      String topicName = args[0].toString();
      Properties props = new Properties();
      
      props.put("bootstrap.servers", "localhost: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.serializa-tion.StringDeserializer");
      props.put("value.deserializer", 
         "org.apache.kafka.common.serializa-tion.StringDeserializer");
      KafkaConsumer<String, String> consumer = new KafkaConsumer
         <String, String>(props);
      
      //Kafka Consumer subscribes list of topics here.
      consumer.subscribe(Arrays.asList(topicName))
      
      //print the topic name
      System.out.println("Subscribed to topic " + topicName);
      int i = 0;
      
      while (true) {
         ConsumerRecords<String, String> records = con-sumer.poll(100);
         for (ConsumerRecord<String, String> record : records)
         
         // print the offset,key and value for the consumer records.
         System.out.printf("offset = %d, key = %s, value = %s\n", 
            record.offset(), record.key(), record.value());
      }
   }
}

Compilation - Ứng dụng có thể được biên dịch bằng lệnh sau.

javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*” *.java

Execution − Ứng dụng có thể được thực thi bằng lệnh sau

java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*”:. SimpleConsumer <topic-name>

Input- Mở CLI của nhà sản xuất và gửi một số thông điệp cho chủ đề. Bạn có thể đặt đầu vào đơn giản là 'Xin chào người tiêu dùng'.

Output - Sau đây sẽ là đầu ra.

Subscribed to topic Hello-Kafka
offset = 3, key = null, value = Hello Consumer