快捷導(dǎo)航

Java使用kafka發(fā)送和生產(chǎn)消息的示例

更新時間：2018年04月24日 15:59:17 作者：will的猜想

本篇文章主要介紹了Java使用kafka發(fā)送和生產(chǎn)消息的示例，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧

1. maven依賴包

<dependency> 
 <groupId>org.apache.kafka</groupId> 
 <artifactId>kafka-clients</artifactId> 
 <version>0.9.0.1</version> 
</dependency>

2. 生產(chǎn)者代碼

package com.lnho.example.kafka;  
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", "master: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<>("topic1", Integer.toString(i), Integer.toString(i)));   
  producer.close(); 
 } 
}

3. 消費者代碼

package com.lnho.example.kafka;   
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", "master: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("topic1")); 
  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()); 
  } 
 } 
}

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章:

相關(guān)文章

Redisson分布式鎖的原理和代碼實例
這篇文章主要介紹了Redisson分布式鎖的原理和代碼實例,在分布式系統(tǒng)中,鎖機(jī)制是非常重要的,Redisson是一個基于Redis的Java應(yīng)用程序,常常被應(yīng)用作為分布式鎖的解決方案,需要的朋友可以參考下
2024-01-01
深入了解Java中Cookie和Session的區(qū)別
會話跟蹤是Web程序中常用的技術(shù),用來跟蹤用戶的整個會話，常用的會話跟蹤技術(shù)是Cookie與Session，本文就詳細(xì)的介紹一下Java中Cookie和Session的區(qū)別，感興趣的可以了解一下
2023-06-06
Mybatis實現(xiàn)自定義的typehandler三步曲
這篇文章主要介紹了Mybatis實現(xiàn)自定義的typehandler三步曲的相關(guān)資料,非常不錯，具有參考借鑒價值，需要的朋友可以參考下
2016-07-07
SpringBoot中注解實現(xiàn)定時任務(wù)的兩種方式
這篇文章主要介紹了SpringBoot中注解實現(xiàn)定時任務(wù)的兩種方式,SpringBoot 定時任務(wù)是一種在SpringBoot應(yīng)用中自動執(zhí)行任務(wù)的機(jī)制,通過使用Spring框架提供的@Scheduled注解,我們可以輕松地創(chuàng)建定時任務(wù),需要的朋友可以參考下
2023-10-10
Java實現(xiàn)數(shù)組翻轉(zhuǎn)的實現(xiàn)代碼
這篇文章主要介紹了Java實現(xiàn)數(shù)組翻轉(zhuǎn)的實現(xiàn)代碼，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧
2019-09-09
Java設(shè)計模式之java中介者模式詳解
這篇文章主要為大家詳細(xì)介紹了23種設(shè)計模式之java中介者模式，具有一定的參考價值，感興趣的小伙伴們可以參考一下,希望能夠給你帶來幫助
2021-09-09
Spring boot webService使用方法解析
這篇文章主要介紹了Spring boot webService使用方法解析,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下
2020-09-09
Java開發(fā)SSM框架微信退款的實現(xiàn)
這篇文章是Java微信退款的教程，退款之前用戶需要先進(jìn)行支付，支付之后才可以使用退款，非常具有實用價值，感興趣的小伙伴們可以參考一下
2018-10-10
劍指Offer之Java算法習(xí)題精講二叉樹專項解析
跟著思路走，之后從簡單題入手，反復(fù)去看，做過之后可能會忘記，之后再做一次，記不住就反復(fù)做，反復(fù)尋求思路和規(guī)律，慢慢積累就會發(fā)現(xiàn)質(zhì)的變化
2022-03-03
SpringBoot使用Redis緩存MySql的方法步驟
本文主要介紹了SpringBoot使用Redis緩存MySql的方法步驟，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2022-02-02