文档章节

kafka java 基本使用

屌丝Lee
 屌丝Lee
发布于 2016/06/23 19:07
字数 553
阅读 1163
收藏 3
点赞 0
评论 0

maven依赖:

<dependency>
	<groupId>org.apache.kafka</groupId>
	<artifactId>kafka_2.11</artifactId>
	<version>0.9.0.0</version>
</dependency>

kafka producer连接,注意import:

import java.util.Properties;
import org.apache.kafka.clients.producer.ProducerRecord;
import config.Configloader;
import config.MsgCreater;
import config.Timer;
import threads.WriteKFKThreads;

public class TestKFKWriteQPS {
	
	private static int count_callback=0;
	private static long timeStart=0;
	private static int count;
	private static int threads;
	private static int length;
	private static String topic;

	public static void main(String[] args) throws Exception {
		
		Properties kp=Configloader.getProperties("../kafka.properties");
		Properties mp=Configloader.getProperties("../main.properties");
		count=Integer.parseInt(mp.getProperty("w_count"));
		threads=Integer.parseInt(mp.getProperty("w_threads"));
		count_callback = threads;
		length=Integer.parseInt(mp.getProperty("msg_len"));
		topic=mp.getProperty("topic");
		
		ProducerRecord<String,String> data = new ProducerRecord<String, String>(topic, MsgCreater.getMsg(length));
		timeStart=System.currentTimeMillis();
		
		for(int i=0;i<threads;i++){
			new Thread(new WriteKFKThreads(count, kp, data)).start();
		}
	}
	
	public static void threadCallback(){
		count_callback--;
		if(count_callback==0){
			long used =System.currentTimeMillis()-timeStart;
			System.out.println("SET --END-- "+count*threads+" time : "+ ((double) used/1000));
			System.out.println("QPS: "+count*threads/((double) used/1000));
			System.out.println(Timer.getAllDelayDistribution());
		}
	}
	
}

kafka set 线程 请无视Timer:

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import config.Timer;
import runner.TestKFKWriteQPS;

public class WriteKFKThreads implements Runnable{

	int count=0;
	Producer<String, String> producer;
	ProducerRecord<String, String> data;
	public WriteKFKThreads(int count , Properties config ,ProducerRecord<String, String> data) {
		this.count=count;
		this.producer =new KafkaProducer<String,String>(config); 
		this.data = data;
	}

	@Override
	public void run() {
		Timer t=new Timer();
		for(int i=0;i<count;i++){
			long start = System.currentTimeMillis();
			try {
					producer.send(data).get();				
					long take =System.currentTimeMillis() - start;
					t.inTime(take);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		Timer.threadAdder(t);
		TestKFKWriteQPS.threadCallback();
		producer.close();
	}
}

kafka consumer连接  :

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 config.Configloader;

public class TestKFKReadQPS {

	private static long timeStart = 0;
	private static int count;
	private static int threads;
	private static String topic;
	private static int total;
	

	public static void main(String[] args) throws Exception {

		Properties kp = Configloader.getProperties("../kafka.properties");
		Properties mp = Configloader.getProperties("../main.properties");
		count = Integer.parseInt(mp.getProperty("r_count"));
		threads = Integer.parseInt(mp.getProperty("r_threads"));
		topic = mp.getProperty("topic");
		total = count * threads;
	//	partition=Integer.parseInt(mp.getProperty("partition"));
	//	offset = Long.parseLong(mp.getProperty("offset"));
		
		KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(kp);
	//	consumer.seek(new TopicPartition(topic, partition), offset);
		


		boolean goon = true;
		consumer.subscribe(Arrays.asList(topic));
		int times=1000;
		while (goon) {
			ConsumerRecords<String, String> records = consumer.poll(100);
			if(records.count() == 0){
				times--;
				if(times <=0){
					goon=false;
					System.out.println("data loose , now get : "+ (count * threads-total));
					break;
				}
			}
			for (ConsumerRecord<String, String> record : records) {
//				 System.out.printf("partition = %d, offset = %d, key = %s,value = %s \n", record.partition(), record.offset(),
//				record.key(), record.value());
				setStart();
				total--;
				if(total <=0){
					goon = false;
					break;

				}
			}
		}
		consumer.close();
		long used =System.currentTimeMillis()-timeStart;
		System.out.println("GET --END-- "+count*threads+" time : "+ ((double) used/1000));
		System.out.println("QPS: "+count*threads/((double) used/1000));

	}
	private static void setStart(){
		if(timeStart == 0){
			timeStart = System.currentTimeMillis();
			System.out.println("Read start!");
		}
	}

}

配置文件 kafka.properties:

group.id=test
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
zookeeper.connect=127.0.0.1:2181
bootstrap.servers=127.0.0.1:9092
metadata.broker.list=127.0.0.1:9092
serializer.class=kafka.serializer.StringEncoder
enable.auto.commit=true
auto.commit.interval.ms=1000
session.timeout.ms=30000
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

main.properties参数很简单,各位自己改改就行。

注:kafka部署时候server是按照linux机器hostname做白名单的(貌似mac下是IP?),也可以设置广播hostname

server.properties中添加:advertised.host.name = XX  

XX最好配置为IP,否则就要在本地host指定 IP =XX

 

 

© 著作权归作者所有

共有 人打赏支持
屌丝Lee
粉丝 1
博文 27
码字总数 11777
作品 0
海淀
请教Kafka在window下运行错误的问题

我在window server 2012 上运行 kafka 的时候 出现这个错误 我本地是可以的,但是在线上服务器就出错了 应该如何解决,请教下 !! Java的版本: C:UsersAdministrator>java -version java ...

Macrotea
04/21
0
0
Zookeeper+Kafka集群搭建

Zookeeper集群搭建 Kafka集群是把状态保存在Zookeeper中的,首先要搭建Zookeeper集群。 1、软件环境 (3台服务器-我的测试) 192.168.30.204 server1 192.168.30.205 server2 192.168.30.206...

qianghong000
06/26
0
0
Windows 安装运行 Apache Kafka 教程

下面是分步指南,教你如何在Windows OS上安装运行Apache Zookeeper和Apache Kafka。 简介 本文讲述了如何在Windows OS上配置并启动Apache Kafka,这篇指南将会指导你安装Java和Apache Zookee...

大数据之路
2012/08/26
0
0
Grafana、elasticsearch、kafka、logstash和pinpoint结合

一、Grafana 1)下载安装 wget https://s3-us-west-2.amazonaws.com/grafana-releases/release/grafana-4.2.0-1.x8664.rpm sudo yum localinstall grafana-4.2.0-1.x86_64.rpm 2)启动 serv......

半船水
2017/10/25
0
0
阿里年薪50WJAVA工程师转大数据学习路线!

大数据有两个方向,一个是偏计算机的,另一个是偏经济的。你学过Java,所以你可以偏将计算机的。 Java程序员想转大数据可行吗?Java是全世界使用人数最多的编程语言。不少程序员选择Java做为...

JAVA丶学习
04/25
0
0
Kafka 0.10.x的Consumer 和 Producer的Java实现

1.Kafka 介绍 Apache Kafka® is a distributed streaming platform. 官网地址 2.maven 依赖 3.手动创建一个 Topic 安装 Kafka 的步骤省略 确保已经安装了Kafka 和 zookeeper(尽量不用Kafka...

移动的红烧肉
05/31
0
0
kafka开源管理工具Kafka-manager部署

  简介      Kafka-manager是雅虎开源的apache-kafka管理工具,是用Scala写,所以在web页面进行操作即可。   Githubhttps://github.com/yahoo/kafka-manager   主要特性:   1. ...

linux运维菜
04/19
0
0
Kafka1.1.0集群的简单使用(java)

Kafka本地集群搭建完成,简单入门Kafka集群的shell操作后,开始Java代码实现简单功能。 kafka版本说明:此处使用最新版本(现在)—1.1.0版本 1.1.0 is the latest release. The current sta...

tian330726
04/19
0
0
加强Docker容器与Java 10集成

很多运行在Java虚拟机(JVM)中的应用,包括数据服务如Apache Spark和Kafka以及传统企业应用,都运行在容器中。最近,运行在容器里的JVM出现了由于内存和CPU资源限制和使用率导致性能损失问题...

java高级架构牛人
06/04
0
0
2.ActiveMQ消息队列安装使用

全程是MOM (Message Oriented Middleware) 消息中间件 消息中间件有很多,比如: 1.ActiveMQ java语言编写的和java系统结合紧密 2.RabbitMQ Erlong语言开发的,天生支持高并发,性能优于A...

小杰java
2017/10/26
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

如何进行大数据的入门级学习?

不知道你是计算机专业应届生还是已经从业者。总之,有java基础的学生学习大数据会轻松很多,零基础的小白都需要从java和linux学起。 如果你是一个学习能力特别强,而且自律性也很强的人的话可...

董黎明
14分钟前
0
0
使用Parcelable传递复杂参数

最近做AIDL传递对象,对象必须实现Parcelable的方法才可以被传递。 @Override    public int describeContents() {//这个 默认返回0就行了。        return 0;    }    ...

火云
15分钟前
0
0
十大Intellij IDEA快捷键

Intellij IDEA中有很多快捷键让人爱不释手,stackoverflow上也有一些有趣的讨论。每个人都有自己的最爱,想排出个理想的榜单还真是困难。以前也整理过Intellij的快捷键,这次就按照我日常开发...

HJCui
25分钟前
0
0
word 使用mathtype 编写 数学公式

下载安装,这个链接命名。。。。 http://www.mathtype.cn/xiazai.html 安装之后会多出一个选项 使用内联方式插入图表 编写公式的界面 设置支持latex 语法 输入公式回车就可以看到结果...

阿豪boy
43分钟前
0
0
Promise

定义 Promise是异步编程的一种解决方案,所谓Promise就是一个容器,里面保存着某个未来才会结束的事件(通常是一个一步操作)的结果。 特点: 2.1 对象的状态不受外界影响,三种状态pending...

litCabbage
54分钟前
1
0
设计模式:适配器模式

说明:在不改变旧接口代码的前提下,为该接口新增其他接口的功能 适配器模式可以分为:类适配器模式、对象适配器模式、接口适配器模式 前两种模式下,我会以播放器为例。老版的播放器(Playe...

人觉非常君
54分钟前
0
0
使用VsCode搭建Java开发环境,创建springboot应用

1、在 Visual Studio Code 中打开扩展视图(Ctrl+Shift+X),输入关键词java、spring分别下载Java开发插件包和springboot插件包 2、配置参数 点击设置按钮,进入设置选项,配置用户设置 在用户...

qsyan
58分钟前
22
0
调教属于你的“贾维斯”(给自己挖了一个很大的坑)

今天玩一下现在很火的人工智能。 废话不多说,先来看几张图: 看出什么蹊跷了吗? 再来看一个视频: https://www.zhihu.com/video/1002567561061511168 (演示网址和代码见文末) 人工智能离...

crossin
今天
1
0
ES10-检索入门

1.创建索引,准备数据 定义索引结构 DELETE telegraphPUT telegraph{ "mappings": { "msg":{ "properties": { "title":{ "type": "text", ...

贾峰uk
今天
1
0
css 三角形

1.无边框三角形 条件:width,height设置为0,border-width设置一定的宽度,border-color保留一边有颜色其他三边设置透明 .triangle{ width:0px; height:0px; border-width:40...

爱喝水的小熊
今天
0
0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部