Java实现Kafka生产者消费者代码实例

Kafka的结构与RabbitMQ类似，消息生产者向Kafka服务器发送消息，Kafka接收消息后，再投递给消费者。
生产者的消费会被发送到Topic中，Topic中保存着各类数据，每一条数据都使用键、值进行保存。
每一个Topic中都包含一个或多个物理分区（Partition），分区维护着消息的内容和索引，它们有可能被保存在不同服务器。

新建一个Maven项目，pom.xml 加入依赖：

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

1、编写生产者

将消息投递到Kafka服务器的名称为“topic1”的Topic中

package com.example.kafkatest;

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

import java.util.Properties;

public class Producer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服务器地址
    props.put("bootstrap.servers", "localhost:9092");
    //设置数据key和value的序列化处理类
    props.put("key.serializer", StringSerializer.class);
    props.put("value.serializer", StringSerializer.class);
    //创建生产者实例
    KafkaProducer<String,String> producer = new KafkaProducer<>(props);
    ProducerRecord record = new ProducerRecord<String, String>("topic1", "userName", "lc");
    //发送记录
    producer.send(record);
    producer.close();
  }
}

运行后，可打开命令行工具，进入Kafka目录，执行命令查询服务器的Topic：

bin\windows\kafka-topics.bat --list --zookeeper localhost:2181

结果如下：

2、编写消费者

本例中，消费者和生产者在同一个项目中，只是使用不同的启动类。

消费者会为自已指定一个消费者组的标识，每一条发布到Topic的记录，都会被交付给消费者组的一个消费者实例。

如果多个消费者实例有相同的消费者组，则这些记录会分配到各个消费者实例上，以达到负载均衡的目录。

如果所有的消费者有不同的消费者组，则每一条记录都会广播到全部的消费者进行处理。

package com.example.rabbittest;

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;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class Consumer {
  public static void main(String[] args) {
    //配置信息
    Properties props = new Properties();
    //kafka服务器地址
    props.put("bootstrap.servers", "localhost:9092");
    //必须指定消费者组
    props.put("group.id", "test");
    //设置数据key和value的序列化处理类
    props.put("key.deserializer", StringDeserializer.class);
    props.put("value.deserializer", StringDeserializer.class);
    //创建消息者实例
    KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
    //订阅topic1的消息
    consumer.subscribe(Arrays.asList("topic1"));
    //到服务器中读取记录
    while (true){
      ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
      for(ConsumerRecord<String,String> record : records){
        System.out.println("key:" + record.key() + "" + ",value:" + record.value());
      }
    }
  }
}

运行后，IDEA控制台其中输出如下：

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

赞 (0)

相关推荐

  • Java使用kafka发送和生产消息的示例

    1. maven依赖包 <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.9.0.1</version> </dependency> 2. 生产者代码 package com.lnho.example.kafka; import org.apache.kafka.c

  • Kafka利用Java实现数据的生产和消费实例教程

    前言 在上一篇中讲述如何搭建kafka集群,本篇则讲述如何简单的使用 kafka .不过在使用kafka的时候,还是应该简单的了解下kafka. Kafka的介绍 Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据. Kafka 有如下特性: 以时间复杂度为O(1)的方式提供消息持久化能力,即使对TB级以上数据也能保证常数时间复杂度的访问性能. 高吞吐率.即使在非常廉价的商用机器上也能做到单机支持每秒100K条以上消息的传输. 支持Kafka Serv

  • Java API方式调用Kafka各种协议的方法

    众所周知,Kafka自己实现了一套二进制协议(binary protocol)用于各种功能的实现,比如发送消息,获取消息,提交位移以及创建topic等.具体协议规范参见:Kafka协议  这套协议的具体使用流程为: 1.客户端创建对应协议的请求 2.客户端发送请求给对应的broker 3.broker处理请求,并发送response给客户端 虽然Kafka提供的大量的脚本工具用于各种功能的实现,但很多时候我们还是希望可以把某些功能以编程的方式嵌入到另一个系统中.这时使用Java API的方式就显

  • Kafka使用Java客户端进行访问的示例代码

    本文环境如下: 操作系统:CentOS 6 32位 JDK版本:1.8.0_77 32位 Kafka版本:0.9.0.1(Scala 2.11) 1. maven依赖包 <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.9.0.1</version> </dependen

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

    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://mave

  • 浅谈Java消息队列总结篇（ActiveMQ、RabbitMQ、ZeroMQ、Kafka）

    一.消息队列概述 消息队列中间件是分布式系统中重要的组件,主要解决应用解耦,异步消息,流量削锋等问题,实现高性能,高可用,可伸缩和最终一致性架构.目前使用较多的消息队列有ActiveMQ,RabbitMQ,ZeroMQ,Kafka,MetaMQ,RocketMQ. 二.消息队列应用场景 以下介绍消息队列在实际应用中常用的使用场景.异步处理,应用解耦,流量削锋和消息通讯四个场景. 2.1异步处理 场景说明:用户注册后,需要发注册邮件和注册短信.传统的做法有两种 1.串行的方式;2.并行方式 a.串

  • Spring纯Java配置集成kafka代码实例

    这篇文章主要介绍了Spring纯Java配置集成kafka代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 KafkaConfig.java package com.niugang.config; import java.util.HashMap; import java.util.Map; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache

  • Java实现Kafka生产者消费者代码实例

    Kafka的结构与RabbitMQ类似,消息生产者向Kafka服务器发送消息,Kafka接收消息后,再投递给消费者. 生产者的消费会被发送到Topic中,Topic中保存着各类数据,每一条数据都使用键.值进行保存. 每一个Topic中都包含一个或多个物理分区(Partition),分区维护着消息的内容和索引,它们有可能被保存在不同服务器. 新建一个Maven项目,pom.xml 加入依赖: <dependency> <groupId>org.apache.kafka</gro

  • JAVA多线程实现生产者消费者的实例详解

    JAVA多线程实现生产者消费者的实例详解 下面的代码实现了生产者消费者的问题 Product.Java package consumerProducer; public class Product { private String id; public String getId() { return id; } public void setId(String id) { this.id = id; } public Product(String id) { this.id=id; } publ

  • Java多线程并发生产者消费者设计模式实例解析

    一.两个线程一个生产者一个消费者 需求情景 两个线程,一个负责生产,一个负责消费,生产者生产一个,消费者消费一个. 涉及问题 同步问题:如何保证同一资源被多个线程并发访问时的完整性.常用的同步方法是采用标记或加锁机制. wait() / nofity() 方法是基类Object的两个方法,也就意味着所有Java类都会拥有这两个方法,这样,我们就可以为任何对象实现同步机制. wait()方法:当缓冲区已满/空时,生产者/消费者线程停止自己的执行,放弃锁,使自己处于等待状态,让其他线程执行. not

  • Java实现Kafka生产者和消费者的示例

    Java实现Kafka生产者和消费者的示例

    Kafka简介 Kafka是由Apache软件基金会开发的一个开源流处理平台,由Scala和Java编写.Kafka的目标是为处理实时数据提供一个统一.高吞吐.低延迟的平台. 方式一:kafka-clients 引入依赖 在pom.xml文件中,引入kafka-clients依赖: <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId

  • java多线程解决生产者消费者问题

    本文实例讲述了java多线程解决生产者消费者问题的方法.分享给大家供大家参考.具体分析如下: 题目是这样的: 采用Java 多线程技术,设计实现一个符合生产者和消费者问题的程序.对一个对象(枪膛)进行操作,其最大容量是12颗子弹.生产者线程是一个压入线程,它不断向枪膛中压入子弹:消费者线程是一个射出线程,它不断从枪膛中射出子弹. 要求: (1)给出分析过程说明. (2)程序输出,要模拟体现对枪膛的压入和射出操作: (2)设计程序时应考虑到两个线程的同步问题. 这个和著名的生产者消费者问题几乎是一

  • java 中多线程生产者消费者问题详细介绍

    java 中多线程生产者消费者问题 前言: 一般面试喜欢问些线程的问题,较基础的问题无非就是死锁,生产者消费者问题,线程同步等等,在前面的文章有写过死锁,这里就说下多生产多消费的问题了 import java.util.concurrent.locks.*; class BoundedBuffer { final Lock lock = new ReentrantLock();//对象锁 final Condition notFull = lock.newCondition(); //生产者监视

  • Queue 实现生产者消费者模型(实例讲解)

    Python中,队列是线程间最常用的交换数据的形式. Python Queue模块有三种队列及构造函数: 1.Python Queue模块的FIFO队列先进先出. class Queue.Queue(maxsize) 2.LIFO类似于堆,即先进后出. class Queue.LifoQueue(maxsize) 3.还有一种是优先级队列级别越低越先出来. class Queue.PriorityQueue(maxsize) 此包中的常用方法(q = Queue.Queue()): q.qsiz

  • Java 读取excel 文件流代码实例

    这篇文章主要介绍了Java 读取excel 文件流代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 代码如下 public static void initType(String path) { try { //1 获取excel文件流 excel xls 文件 暂不支持xlsx if (path.contains("xlsx") || path.contains("XLSX")) { System.err.

  • java实现python session功能代码实例

    这篇文章主要介绍了java实现python session功能代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 怎么在java中实现类似于python的requests模块的session功能呢.java也是可以实现的,用java的okhttp包可以实现. 在pom.xml中加入相关依赖 <!-- https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp --> <

  • java io读取文件操作代码实例

    这篇文章主要介绍了java io读取文件操作代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下 主要分为字节读取和字符读取,字节读取可以一个一个读取和字节数组读取,字符读取同样之,字符读取适合文本读取,字节读取皆可以 这里直接上代码,读取文件的9个小demo package com.io; import org.junit.Test; import java.io.*; public class FileTest { //1.字节流字节一个

随机推荐