文档章节

kafka java 基本使用

屌丝Lee
 屌丝Lee
发布于 2016/06/23 19:07
字数 553
阅读 1173
收藏 3

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

没有更多内容

加载失败,请刷新页面

加载更多

jQuery学习笔记180923

射手

颖伙虫
8分钟前
0
0
[python] colorama 模块 - 改变控制台输出文本的颜色

除了使用 PyQt 这样的图形化开发框架外,基本上 python 程序都是跑在控制台中的。很多时候,单纯使用黑白的文字不能很好地突出我们要显示的信息。有时候我们需要将错误的提示使用红色标注,而...

cometeme
13分钟前
0
0
Makefile 学习 2 - 基于若干 Blog 的汇总

基于若干 Blog 汇总的 makefile 教程 陈皓 https://blog.csdn.net/haoel/article/details/2886 Makefile 进阶 1. Makefile 中的内容 显式规则。显式规则说明了,如何生成一个或多的的目标文件...

公孙衍
29分钟前
0
0
NIO与BIO的区别、NIO的运行原理和并发使用场景

NIO(Non-blocking I/O,在Java领域,也称为New I/O),是一种同步非阻塞的I/O模型,也是I/O多路复用的基础,已经被越来越多地应用到大型应用服务器,成为解决高并发与大量连接、I/O处理问题的...

Java干货分享
46分钟前
1
0
72.告警系统邮件引擎 运行告警系统

20.23/20.24/20.25 告警系统邮件引擎 20.26 运行告警系统 20.23/20.24/20.25 告警系统邮件引擎 邮件首先要有一个mail.py,以下。 因为我们之前zabbix的时候做过,就可以直接拷贝过来 mail.s...

王鑫linux
今天
1
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部