脚本之家 服务器常用软件
快捷导航
您的位置：首页 → 软件编程 → java → java kafka 生产消费

kafka生产者和消费者的javaAPI的示例代码

 更新时间：2018年06月18日 11:52:37   作者：lijie_cq  
这篇文章主要介绍了kafka生产者和消费者的javaAPI的示例代码，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

写了个kafka的java demo 顺便记录下，仅供参考

1.创建maven项目

目录如下：

2.pom文件：

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>Kafka-Maven</groupId>
  <artifactId>Kafka-Maven</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <dependencies>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka_2.11</artifactId>
      <version>0.10.1.1</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-common</artifactId>
      <version>2.2.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-hdfs</artifactId>
      <version>2.2.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client</artifactId>
      <version>2.2.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-client</artifactId>
      <version>1.0.3</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hbase</groupId>
      <artifactId>hbase-server</artifactId>
      <version>1.0.3</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-hdfs</artifactId>
      <version>2.2.0</version>
    </dependency>
    <dependency>
      <groupId>jdk.tools</groupId>
      <artifactId>jdk.tools</artifactId>
      <version>1.7</version>
      <scope>system</scope>
      <systemPath>${JAVA_HOME}/lib/tools.jar</systemPath>
    </dependency>
    <dependency>
      <groupId>org.apache.httpcomponents</groupId>
      <artifactId>httpclient</artifactId>
      <version>4.3.6</version>
    </dependency>
  </dependencies>
  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <configuration>
          <source>1.7</source>
          <target>1.7</target>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>

3.kafka生产者KafkaProduce:

package com.lijie.producer;

import java.io.File;
import java.io.FileInputStream;
import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class KafkaProduce {
  private static Properties properties;

  static {
    properties = new Properties();
    String path = KafkaProducer.class.getResource("/").getFile().toString()
        + "kafka.properties";
    try {
      FileInputStream fis = new FileInputStream(new File(path));
      properties.load(fis);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /**
   * 发送消息
   * 
   * @param topic
   * @param key
   * @param value
   */
  public void sendMsg(String topic, byte[] key, byte[] value) {

    // 实例化produce
    KafkaProducer<byte[], byte[]> kp = new KafkaProducer<byte[], byte[]>(
        properties);

    // 消息封装
    ProducerRecord<byte[], byte[]> pr = new ProducerRecord<byte[], byte[]>(
        topic, key, value);

    // 发送数据
    kp.send(pr, new Callback() {
      // 回调函数
      @Override
      public void onCompletion(RecordMetadata metadata,
          Exception exception) {
        if (null != exception) {
          System.out.println("记录的offset在:" + metadata.offset());
          System.out.println(exception.getMessage() + exception);
        }
      }
    });

    // 关闭produce
    kp.close();
  }
}

4.kafka消费者KafkaConsume：

package com.lijie.consumer;

import java.io.File;
import java.io.FileInputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.htrace.fasterxml.jackson.databind.ObjectMapper;

import com.lijie.pojo.User;
import com.lijie.utils.JsonUtils;

import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;

public class KafkaConsume {

  private final static String TOPIC = "lijietest";

  private static Properties properties;

  static {
    properties = new Properties();
    String path = KafkaConsume.class.getResource("/").getFile().toString()
        + "kafka.properties";
    try {
      FileInputStream fis = new FileInputStream(new File(path));
      properties.load(fis);
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  /**
   * 获取消息
   * 
   * @throws Exception
   */
  public void getMsg() throws Exception {
    ConsumerConfig config = new ConsumerConfig(properties);

    ConsumerConnector consumer = kafka.consumer.Consumer
        .createJavaConsumerConnector(config);

    Map<String, Integer> topicCountMap = new HashMap<String, Integer>();

    topicCountMap.put(TOPIC, new Integer(1));

    StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());

    StringDecoder valueDecoder = new StringDecoder(
        new VerifiableProperties());

    Map<String, List<KafkaStream<String, String>>> consumerMap = consumer
        .createMessageStreams(topicCountMap, keyDecoder, valueDecoder);

    KafkaStream<String, String> stream = consumerMap.get(TOPIC).get(0);

    ConsumerIterator<String, String> it = stream.iterator();

    while (it.hasNext()) {
      String json = it.next().message();
      User user = (User) JsonUtils.JsonToObj(json, User.class);
      System.out.println(user);
    }
  }
}

5.kafka.properties文件

##produce
bootstrap.servers=192.168.80.123:9092
producer.type=sync
request.required.acks=1
serializer.class=kafka.serializer.DefaultEncoder
key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer
bak.partitioner.class=kafka.producer.DefaultPartitioner
bak.key.serializer=org.apache.kafka.common.serialization.StringSerializer
bak.value.serializer=org.apache.kafka.common.serialization.StringSerializer

##consume
zookeeper.connect=192.168.80.123:2181 
group.id=lijiegroup 
zookeeper.session.timeout.ms=4000 
zookeeper.sync.time.ms=200 
auto.commit.interval.ms=1000 
auto.offset.reset=smallest 
serializer.class=kafka.serializer.StringEncoder

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。

您可能感兴趣的文章:

相关文章

  • java实现图片转ascii字符画的方法示例

    java实现图片转ascii字符画的方法示例

    这篇文章主要介绍了java实现图片转ascii字符画的方法示例，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2018-08-08
  • java读写excel文件实现POI解析Excel的方法

    java读写excel文件实现POI解析Excel的方法

    在日常工作中，我们常常会进行Excel文件读写操作，这篇文章主要介绍了java读写excel文件实现POI解析Excel的方法,实例分析了java读写excel的技巧，非常具有实用价值，需要的朋友可以参考下
    2018-10-10
  • 浅谈JVM之使用JFR解决内存泄露

    浅谈JVM之使用JFR解决内存泄露

    内存泄露的主要原因就是java中的对象生命周期有长有短。如果长生命周期的对象引用了短生命周期的对象，就有可能造成事实上的内存泄露。本文将介绍JVM之使用JFR解决内存泄露。
    2021-06-06
  • Swagger-boostrap-ui如何配置用户名密码访问

    Swagger-boostrap-ui如何配置用户名密码访问

    这篇文章主要介绍了Swagger-boostrap-ui如何配置用户名密码访问，具有很好的参考价值，希望对大家有所帮助。如有错误或未考虑完全的地方，望不吝赐教
    2023-05-05
  • java编程中字节流转换成字符流的实现方法

    java编程中字节流转换成字符流的实现方法

    下面小编就为大家带来一篇java编程中字节流转换成字符流的实现方法。小编觉得挺不错的，现在就分享给大家，也给大家做个参考。一起跟随小编过来看看吧
    2017-01-01
  • 解决SpringBoot运行报错：找不到或无法加载主类的问题

    解决SpringBoot运行报错：找不到或无法加载主类的问题

    这篇文章主要介绍了解决SpringBoot运行报错：找不到或无法加载主类的问题,具有很好的参考价值,对大家的学习或工作有一定的参考价值,需要的朋友可以参考下
    2023-09-09
  • IDEA创建parent项目(聚合项目)

    IDEA创建parent项目(聚合项目)

    这篇文章主要介绍了IDEA创建parent项目(聚合项目),文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
    2019-08-08
  • Spring如何解决循环依赖的问题

    Spring如何解决循环依赖的问题

    这篇文章主要介绍了Spring是如何解决循环依赖的问题，本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
    2020-08-08
  • Java源码解析之TypeVariable详解

    Java源码解析之TypeVariable详解

    这篇文章主要介绍了Java源码解析之TypeVariable详解，具有一定参考价值，需要的朋友可以了解下。
    2017-10-10
  • spring中bean的生命周期详解

    spring中bean的生命周期详解

    今天小编就为大家分享一篇关于spring中bean的生命周期详解，小编觉得内容挺不错的，现在分享给大家，具有很好的参考价值，需要的朋友一起跟随小编来看看吧
    2019-01-01

最新评论