Kafka 使用Java实现数据的生产和消费demo

前言

在上一篇中讲述如何搭建kafka集群，本篇则讲述如何简单的使用 kafka 。不过在使用kafka的时候，还是应该简单的了解下kafka。

Kafka的介绍

Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者规模的网站中的所有动作流数据。
Kafka 有如下特性：

  • 以时间复杂度为O(1)的方式提供消息持久化能力，即使对TB级以上数据也能保证常数时间复杂度的访问性能。
  • 高吞吐率。即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息的传输。
  • 支持Kafka Server间的消息分区，及分布式消费，同时保证每个Partition内的消息顺序传输。
  • 同时支持离线数据处理和实时数据处理。
  • Scale out：支持在线水平扩展。

kafka的术语

  • Broker：Kafka集群包含一个或多个服务器，这种服务器被称为broker。
  • Topic：每条发布到Kafka集群的消息都有一个类别，这个类别被称为Topic。（物理上不同Topic的消息分开存储，逻辑上一个Topic的消息虽然保存于一个或多个broker上但用户只需指定消息的Topic即可生产或消费数据而不必关心数据存于何处）
  • Partition：Partition是物理上的概念，每个Topic包含一个或多个Partition。
  • Producer：负责发布消息到Kafka broker。
  • Consumer：消息消费者，向Kafka broker读取消息的客户端。
  • Consumer Group:每个Consumer属于一个特定的Consumer Group（可为每个Consumer指定group name，若不指定group name则属于默认的group）。

kafka核心Api

kafka有四个核心API

  • 应用程序使用producer API发布消息到1个或多个topic中。
  • 应用程序使用consumer API来订阅一个或多个topic,并处理产生的消息。
  • 应用程序使用streams API充当一个流处理器,从1个或多个topic消费输入流,并产生一个输出流到1个或多个topic,有效地将输入流转换到输出流。
  • connector API允许构建或运行可重复使用的生产者或消费者,将topic链接到现有的应用程序或数据系统。

示例图如下:

kafka 应用场景

  • 构建可在系统或应用程序之间可靠获取数据的实时流数据管道。
  • 构建实时流应用程序，可以转换或响应数据流。

以上介绍参考kafka官方文档。

开发准备

如果我们要开发一个kafka的程序，应该做些什么呢？
首先，在搭建好kafka环境之后，我们要考虑的是我们是生产者还是消费者，也就是消息的发送者还是接受者。
不过在本篇中，生产者和消费者都会进行开发和讲解。

在大致的了解kafka之后，我们来开发第一个程序。
这里用的开发语言是Java，构建工具Maven。
Maven的依赖如下:

    <dependency>
        <groupId>org.apache.kafka</groupId>
         <artifactId>kafka_2.12</artifactId>
         <version>1.0.0</version>
            <scope>provided</scope>
        </dependency>

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

        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-streams</artifactId>
            <version>1.0.0</version>
        </dependency>

Kafka Producer

在开发生产的时候，先简单的介绍下kafka各种配置说明：

  • bootstrap.servers： kafka的地址。
  • acks:消息的确认机制，默认值是0。
    acks=0：如果设置为0，生产者不会等待kafka的响应。
    acks=1：这个配置意味着kafka会把这条消息写到本地日志文件中，但是不会等待集群中其他机器的成功响应。
    acks=all：这个配置意味着leader会等待所有的follower同步完成。这个确保消息不会丢失，除非kafka集群中所有机器挂掉。这是最强的可用性保证。
  • retries：配置为大于0的值的话，客户端会在消息发送失败时重新发送。
  • batch.size:当多条消息需要发送到同一个分区时，生产者会尝试合并网络请求。这会提高client和生产者的效率。
  • key.serializer: 键序列化，默认org.apache.kafka.common.serialization.StringDeserializer。
  • value.deserializer:值序列化，默认org.apache.kafka.common.serialization.StringDeserializer。
    ...
    还有更多配置，可以去查看官方文档，这里就不在说明了。
    那么我们kafka 的producer配置如下:
        Properties props = new Properties();
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

kafka的配置添加之后，我们便开始生产数据，生产数据代码只需如下就行:

 producer.send(new ProducerRecord<String, String>(topic,key,value));
  • topic: 消息队列的名称，可以先行在kafka服务中进行创建。如果kafka中并未创建该topic，那么便会自动创建！
  • key:键值，也就是value对应的值，和Map类似。
  • value:要发送的数据，数据格式为String类型的。

在写好生产者程序之后，那我们先来生产吧！
我这里发送的消息为:

 String messageStr="你好，这是第"+messageNo+"条数据";

并且只发送1000条就退出，结果如下:

可以看到信息成功的打印了。
如果不想用程序进行验证程序是否发送成功，以及消息发送的准确性，可以在kafka服务器上使用命令查看。

Kafka Consumer

kafka消费这块应该来说是重点，毕竟大部分的时候，我们主要使用的是将数据进行消费。

kafka消费的配置如下:

  • bootstrap.servers： kafka的地址。
  • group.id：组名 不同组名可以重复消费。例如你先使用了组名A消费了kafka的1000条数据，但是你还想再次进行消费这1000条数据，并且不想重新去产生，那么这里你只需要更改组名就可以重复消费了。
  • enable.auto.commit：是否自动提交，默认为true。
  • auto.commit.interval.ms: 从poll(拉)的回话处理时长。
  • session.timeout.ms:超时时间。
  • max.poll.records:一次最大拉取的条数。
  • auto.offset.reset：消费规则，默认earliest 。
    earliest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 。
    latest: 当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 。
    none: topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常。
  • key.serializer: 键序列化，默认org.apache.kafka.common.serialization.StringDeserializer。
  • value.deserializer:值序列化，默认org.apache.kafka.common.serialization.StringDeserializer。

那么我们kafka 的consumer配置如下:

    Properties props = new Properties();
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        props.put("group.id", GROUPID);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("max.poll.records", 1000);
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

由于我这是设置的自动提交，所以消费代码如下:
我们需要先订阅一个topic，也就是指定消费哪一个topic。

consumer.subscribe(Arrays.asList(topic));

订阅之后，我们再从kafka中拉取数据:

ConsumerRecords<String, String> msgList=consumer.poll(1000);

一般来说进行消费会使用监听，这里我们就用for(;;)来进行监听， 并且设置消费1000条就退出！
结果如下:

可以看到我们这里已经成功消费了生产的数据了。

代码

那么生产者和消费者的代码如下:

生产者:

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 *
* Title: KafkaProducerTest
* Description:
* kafka 生产者demo
* Version:1.0.0
* @author pancm
* @date 2018年1月26日
 */
public class KafkaProducerTest implements Runnable {

    private final KafkaProducer<String, String> producer;
    private final String topic;
    public KafkaProducerTest(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        this.producer = new KafkaProducer<String, String>(props);
        this.topic = topicName;
    }

    @Override
    public void run() {
        int messageNo = 1;
        try {
            for(;;) {
                String messageStr="你好，这是第"+messageNo+"条数据";
                producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
                //生产了100条就打印
                if(messageNo%100==0){
                    System.out.println("发送的信息:" + messageStr);
                }
                //生产1000条就退出
                if(messageNo%1000==0){
                    System.out.println("成功发送了"+messageNo+"条");
                    break;
                }
                messageNo++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    public static void main(String args[]) {
        KafkaProducerTest test = new KafkaProducerTest("KAFKA_TEST");
        Thread thread = new Thread(test);
        thread.start();
    }
}

消费者:

import java.util.Arrays;
import java.util.Properties;
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.serialization.StringDeserializer;

/**
 *
* Title: KafkaConsumerTest
* Description:
*  kafka消费者 demo
* Version:1.0.0
* @author pancm
* @date 2018年1月26日
 */
public class KafkaConsumerTest implements Runnable {

    private final KafkaConsumer<String, String> consumer;
    private ConsumerRecords<String, String> msgList;
    private final String topic;
    private static final String GROUPID = "groupA";

    public KafkaConsumerTest(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        props.put("group.id", GROUPID);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        this.topic = topicName;
        this.consumer.subscribe(Arrays.asList(topic));
    }

    @Override
    public void run() {
        int messageNo = 1;
        System.out.println("---------开始消费---------");
        try {
            for (;;) {
                    msgList = consumer.poll(1000);
                    if(null!=msgList&&msgList.count()>0){
                    for (ConsumerRecord<String, String> record : msgList) {
                        //消费100条就打印 ,但打印的数据不一定是这个规律的
                        if(messageNo%100==0){
                            System.out.println(messageNo+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
                        }
                        //当消费了1000条就退出
                        if(messageNo%1000==0){
                            break;
                        }
                        messageNo++;
                    }
                }else{
                    Thread.sleep(1000);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
    public static void main(String args[]) {
        KafkaConsumerTest test1 = new KafkaConsumerTest("KAFKA_TEST");
        Thread thread1 = new Thread(test1);
        thread1.start();
    }
}

注: master、slave1、slave2 是因为我在自己的环境做了关系映射，这个可以换成服务器的IP。

当然项目我放在Github上了，有兴趣的可以看看。 https://github.com/xuwujing/kafka

总结

简单的开发一个kafka的程序需要以下步骤:

  1. 成功搭建kafka服务器，并成功启动！
  2. 得到kafka服务信息，然后在代码中进行相应的配置。
  3. 配置完成之后，监听kafka中的消息队列是否有消息产生。
  4. 将产生的数据进行业务逻辑处理！

kafka介绍参考官方文档:
http://kafka.apache.org/intro

到此，本文就结束了，谢谢阅读！

原文地址：https://www.cnblogs.com/xuwujing/p/8371127.html

时间： 2024-11-05 23:27:18

Kafka 使用Java实现数据的生产和消费demo的相关文章

JAVA代码之RocketMQ生产和消费数据

一.启动RocketMQ [[email protected] ~]# cat /etc/hosts # Do not remove the following line, or various programs # that require network functionality will fail. 127.0.0.1               localhost.localdomain localhost ::1             localhost6.localdomain6

spring整合kafka项目生产和消费测试结果记录（一）

使用spring+springMVC+mybatis+kafka做了两个web项目,一个是生产者,一个是消费者. 通过JMeter测试工具模拟100个用户并发访问生产者项目,发送json数据给生产者的接口,生产者将json数据发送到kafka集群, 消费者监听到kafka集群中的消息就开始消费,并将json解析成对象存到MySQL数据库. 下面是使用JMeter测试工具模拟100个并发的线程设置截图: 请求所发送的数据: 下面是100个用户10000个请求的聚合报告: 下面是生产者截图生产完10

【sparkStreaming】kafka作为数据源的生产和消费

1.建立生产者发送数据 (1)配置zookeeper属性信息props (2)通过 new KafkaProducer[KeyType,ValueType](props) 建立producer (3)通过 new ProducerRecord[KeyType,ValueType](topic,key,value) 封装消息message (4)通过 producer.send(message) 发送消息 package SparkDemo import java.util import org.

Java多线程学习笔记--生产消费者模式

实际开发中,我们经常会接触到生产消费者模型,如:Android的Looper相应handler处理UI操作,Socket通信的响应过程.数据缓冲区在文件读写应用等.强大的模型框架,鉴于本人水平有限目前水平只能膜拜,本次只能算学习笔记,为了巩固自己对Java多线程常规知识点的理解,路过大神还望能指导指导.下面一段代码是最常规的生产者消费者的例子: package com.zhanglei.demo; import java.util.ArrayList; import java.util.List

Java大数据人才应用领域广，就业薪酬高

互联网创造了大数据应用的规模化环境,大数据应用成功的案例大都是在互联网上发生的, 互联网业务提供了数据,互联网企业开发了处理软件,互联网企业的创新带来了大数据应用 的活跃,没有互联网便没有今天的大数据产业.没有互联网.云计算.物联网.移动终端与 人工智能组合的环境大数据也没那么重要.大数据的价值并非与生俱来而是应用创新之结果 ,价值是由技术组合创新涌现出来的.离开环境的支持大数据毫无价值,就像离开了身体的 手不再有手的功能一样. 随着2017年大数据各种应用的发展,大数据的价值得以充分的发挥,大

4 kafka集群部署及生产者java客户端编程 + kafka消费者java客户端编程

本博文的主要内容有   kafka的单机模式部署 kafka的分布式模式部署 生产者java客户端编程 消费者java客户端编程 运行kafka ,需要依赖 zookeeper,你可以使用已有的 zookeeper 集群或者利用 kafka自带的zookeeper. 单机模式,用的是kafka自带的zookeeper, 分布式模式,用的是外部安装的zookeeper,即公共的zookeeper. Step 6: Setting up a multi-broker cluster So far w

Java的多线程实现生产/消费模式

Java的多线程实现生产/消费模式 在Java的多线程中,我们经常使用某个Java对象的wait(),notify()以及notifyAll() 方法实现多线程的通讯,今天就使用Java的多线程实现生产/消费模式,需求如下: 线程A ProductThread 继承Thread 实现生产数据 若线程共享的数据不为NULL,则生产线程进入等待状态 线程B CustomThread 继承Thread 实现消费数据(输出到控制台) 当线程共享数据为NULL的时候,进入等待状态 线程B 消费完数据之后,

Kafka对Java程序员有多重要？连阿里都再用它处理亿万级数据统计

一．了解淘宝Kafka架构在ActiveMQ.RabbitMQ.RocketMQ.Kafka消息中间件之间,我们为什么要选择Kafka?下面详细介绍一下,2012年9月份我在支付宝做余额宝研发,2013年6月支付宝正式推出余额宝,2013年8月担任支付宝淘宝×××项目经理带领兄弟们一起做研发,期间需要与淘宝和500万对接竞彩接口数据,业余时间与淘宝的同事沟通,了解天猫在电商节如何处理这些大数据的?技术架构上采用了哪些策略呢? 一.应用无状态(淘宝session框架) 二.有效使用缓存(Tair)

Python 基于Python结合pykafka实现kafka生产及消费速率&amp;主题分区偏移实时监控

基于Python结合pykafka实现kafka生产及消费速率&主题分区偏移实时监控   By: 授客 QQ:1033553122   1．测试环境 python 3.4 zookeeper-3.4.13.tar.gz 下载地址1: http://zookeeper.apache.org/releases.html#download https://www.apache.org/dyn/closer.cgi/zookeeper/ https://mirrors.tuna.tsinghua.edu