文档章节

kafka的单机部署版本

l
 lry77
发布于 2016/08/16 20:13
字数 716
阅读 291
收藏 7

本部署使用的版本为kafka_2.8.0-0.8.0。 
参考了http://blog.csdn.net/itleochen/article/details/17451455这篇博文; 
并根据官网介绍http://kafka.apache.org/documentation.html#quickstart完成。 
废话少说,直接上步骤 
1.下载kafka_2.8.0-0.8.0.tar.gz 
https://archive.apache.org/dist/kafka/0.8.0/kafka_2.8.0-0.8.0.tar.gz 
2.解压缩 
tar -vxf kafka_2.8.0-0.8.0.tar.gz 
3.修改配置文件 
修改conf/server.properties 
host.name=192.168.110.129(修改为主机ip,不然服务器返回给客户端的是主机的hostname,客户端并不一定能够识别) 
修改conf/zookeeper.properties 属性文件 
dataDir=/usr/local/tmp/zookeeper   (zookeeper临时数据文件) 
4.启动zookeeper和kafka 
cd bin 
启动zookeeper 
./zookeeper-server-start.sh ../config/zookeeper.properties & (&推出命令行,服务守护执行) 
启动kafka 
./kafka-server-start.sh ../config/server.properties & 
5.验证是否成功 
*创建主题 
./kafka-create-topic.sh --partition 1 --replica 1 --zookeeper localhost:2181 --topic test 
检查是否创建主题成功 
./kafka-list-topic.sh --zookeeper localhost:2181 
*启动produce 
./bin/kafka-console-producer.sh --broker-list 192.168.110.129:9092  --topic test 
*启动consumer 
./kafka-console-consumer.sh --zookeeper localhost:2181 --topic test 
6.关闭kafka和zookeeper 
./kafka-server-stop.sh ../config/server.properties 
./zookeeper-server-stop.sh 
心得总结: 
1.produce启动的时候参数使用的是kafka的端口而consumer启动的时候使用的是zookeeper的端口; 
2.必须先创建topic才能使用; 
3.topic本质是以文件的形式储存在zookeeper上的。

 

消费者

package com.kafka;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

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 KafkaConsumer
{

	private final ConsumerConnector consumer;


	private KafkaConsumer()
	{
		Properties props = new Properties();
		// zookeeper 配置
		props.put( "zookeeper.connect", "192.168.110.129:2181" );

		// group 代表一个消费组
		props.put( "group.id", "jd-group" );

		// zk连接超时
		props.put( "zookeeper.session.timeout.ms", "4000" );
		props.put( "zookeeper.sync.time.ms", "200" );
		props.put( "auto.commit.interval.ms", "1000" );
		props.put( "auto.offset.reset", "smallest" );
		// 序列化类
		props.put( "serializer.class", "kafka.serializer.StringEncoder" );

		ConsumerConfig config = new ConsumerConfig( props );

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


	void consume()
	{
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
		topicCountMap.put( KafkaProducer.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( KafkaProducer.TOPIC ).get( 0 );
		ConsumerIterator<String, String> it = stream.iterator();
		while (it.hasNext())
			System.out.println( it.next().message() );
	}


	public static void main(String[] args)
	{
		new KafkaConsumer().consume();
	}
}

生产者

package com.kafka;

import java.util.Properties;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

/**
 * Hello world!
 *
 */
public class KafkaProducer
{
	private final Producer<String, String> producer;

	public final static String TOPIC = "TEST-TOPIC";


	private KafkaProducer()
	{
		Properties props = new Properties();
		// 此处配置的是kafka的端口
		props.put( "metadata.broker.list", "192.168.110.129:9092" );

		// 配置value的序列化类
		props.put( "serializer.class", "kafka.serializer.StringEncoder" );
		// 配置key的序列化类
		props.put( "key.serializer.class", "kafka.serializer.StringEncoder" );

		// request.required.acks
		// 0, which means that the producer never waits for an acknowledgement from the broker (the same
		// behavior as 0.7). This option provides the lowest latency but the weakest durability guarantees
		// (some data will be lost when a server fails).
		// 1, which means that the producer gets an acknowledgement after the leader replica has received the
		// data. This option provides better durability as the client waits until the server acknowledges the
		// request as successful (only messages that were written to the now-dead leader but not yet
		// replicated will be lost).
		// -1, which means that the producer gets an acknowledgement after all in-sync replicas have received
		// the data. This option provides the best durability, we guarantee that no messages will be lost as
		// long as at least one in sync replica remains.
		props.put( "request.required.acks", "-1" );

		producer = new Producer<String, String>( new ProducerConfig( props ) );
	}


	void produce()
	{
		int messageNo = 1000;
		final int COUNT = 2000;

		while (messageNo < COUNT)
		{
			String key = String.valueOf( messageNo );
			String data = "hello kafka message " + key;
			producer.send( new KeyedMessage<String, String>( TOPIC, key, data ) );
			// System.out.println( data );
			messageNo++;
		}
	}


	public static void main(String[] args)
	{
		new KafkaProducer().produce();
	}
}

 

© 著作权归作者所有

l
粉丝 1
博文 135
码字总数 88749
作品 0
南京
私信 提问
集群四部曲(四):完美的Kafka集群搭建

之前写过一篇关于Kafka消息的发布-订阅,只不过是基于一台服务器,不够全面,下面我要说下Kafka集群环境的搭建和消息的发布-订阅,希望大家喜欢。下面的集群搭建是基于单机部署的环境,所以大...

海岸线的曙光
2018/03/20
260
0
Kafka分布式环境搭建 (二)赞

这篇文章将介绍如何搭建kafka环境,我们会从单机版开始,然后逐渐往分布式扩展。单机版的搭建官网上就有,比较容易实现,这里我就简单介绍下即可,而分布式的搭建官网却没有描述,我们最终的...

老先生二号
2017/08/07
0
0
RocketMQ与Kafka对比

RocketMQ与Kafka对比(18项差异) 淘宝内部的交易系统使用了淘宝自主研发的Notify消息中间件,使用Mysql作为消息存储媒介,可完全水平扩容,为了进一步降低成本,我们认为存储部分可以进一步...

莫问viva
2015/05/08
611
0
RocketMQ与Kafka对比(18项差异)

转自:https://github.com/alibaba/RocketMQ/wiki/rmqvskafka 淘宝内部的交易系统使用了淘宝自主研发的Notify消息中间件,使用Mysql作为消息存储媒介,可完全水平扩容,为了进一步降低成本,...

洋哥6
2016/02/29
238
0
Apache RocketMQ QuickStart

RocketMQ作为一款分布式的消息中间件(阿里的说法是不遵循任何规范的,所以不能完全用JMS的那一套东西来看它),经历了Metaq1.x、Metaq2.x的发展和淘宝双十一的洗礼,在功能和性能上远超Act...

程序员诗人
2017/09/29
0
0

没有更多内容

加载失败,请刷新页面

加载更多

nginx学习笔记

中间件位于客户机/ 服务器的操作系统之上,管理计算机资源和网络通讯。 是连接两个独立应用程序或独立系统的软件。 web请求通过中间件可以直接调用操作系统,也可以经过中间件把请求分发到多...

码农实战
今天
5
0
Spring Security 实战干货:玩转自定义登录

1. 前言 前面的关于 Spring Security 相关的文章只是一个预热。为了接下来更好的实战,如果你错过了请从 Spring Security 实战系列 开始。安全访问的第一步就是认证(Authentication),认证...

码农小胖哥
今天
11
0
JAVA 实现雪花算法生成唯一订单号工具类

import lombok.SneakyThrows;import lombok.extern.slf4j.Slf4j;import java.util.Calendar;/** * Default distributed primary key generator. * * <p> * Use snowflake......

huangkejie
昨天
12
0
PhotoShop 色调:RGB/CMYK 颜色模式

一·、 RGB : 三原色:红绿蓝 1.通道:通道中的红绿蓝通道分别对应的是红绿蓝三种原色(RGB)的显示范围 1.差值模式能模拟三种原色叠加之后的效果 2.添加-颜色曲线:调整图像RGB颜色----R色增强...

东方墨天
昨天
11
1
将博客搬至CSDN

将博客搬至CSDN

算法与编程之美
昨天
13
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部