kafka消息中间件技术安装教程

一、简介：

http://kafka.apache.org 

   kafka是一个分布式消息系统，由linkedin公司使用scala编写，用作LinkedIn的活动流（Activity Stream）和运营数据处理管道（Pipeline）的基础。具有高水平扩展和高吞吐量。

二、kafka的特点：

1、同时为发布和订阅提供高吞吐量。据了解，Kafka每秒可以生产约 25 万消息（50 MB），每秒处理 55 万消息（110 MB）。

2、可进行持久化操作。将消息持久化到磁盘，因此可用于批量消费，例如 ETL，以及实时应用程序。通过将数据持久化到硬盘以及 replication 防止数据丢失。

3、分布式系统，易于向外扩展。所有的producer、broker 和 consumer 都会有多个，均为分布式的。无需停机即可扩展机器。

4、消息被处理的状态是在 consumer 端维护，而不是由server 端维护。当失败时能自动平衡。

5、支持 online 和 offline 的场景。

三、应用场景

1、Messaging  

    对于一些常规的消息系统,kafka是个不错的选择;partitons/replication和容错,可以使kafka具有良好的扩展性和性能优势.不过到目前为止,我们应该很清楚认识到,kafka并没有提供JMS中的"事务性""消息传输担保(消息确认机制)""消息分组"等企业级特性;kafka只能使用作为"常规"的消息系统,在一定程度上,尚未确保消息的发送与接收绝对可靠(比如,消息重发,消息发送丢失等)

2、Website activitytracking

    kafka可以作为"网站活性跟踪"的最佳工具;可以将网页/用户操作等信息发送到kafka中.并实时监控,或者离线统计分析等

3、Log Aggregation

   kafka的特性决定它非常适合作为"日志收集中心";application可以将操作日志"批量""异步"的发送到kafka集群中,而不是保存在本地或者DB中;kafka可以批量提交消息/压缩消息等,这对producer端而言,几乎感觉不到性能的开支.此时consumer端可以使hadoop等其他系统化的存储和分析系统.

四、zookeeper集群搭建：

         #下载zookeeper软件  

         wgethttp://mirrors.cnnic.cn/apache/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz

         #解压软件

         tar-zxvf zookeeper-3.4.6.tar.gz

         #进入到解压好的目录里面的conf目录中，添加一个zoo.cfg配置文件

         mvzoo_sample.cfg zoo.cfg

         #修改配置文件（zoo.cfg）

         dataDir=/it/zookeeper-3.4.6/data(创建data目录)

         server.5=192.168.177.5:2888:3888

         server.6=192.168.177.6:2888:3888

         server.7=192.168.177.7:2888:3888

         #在(dataDir=/it/zookeeper-3.4.6/data)创建一个myid文件，里面内容是server.N中的N（server.2里面内容为2）

            echo"5" > myid

         #将配置好的zk拷贝到其他节点

         scp -r  /it/zookeeper-3.4.6/  it06@192.168.117.6:/itcast/

         scp -r  /it/zookeeper-3.4.6/  it07@192.168.117.7:/itcast/

         #注意：在其他节点上一定要修改myid的内容

                   在it06应该将myid的内容改为6 （echo"6" > myid）

                   在it07应该将myid的内容改为7 （echo"7" > myid）

         #启动集群

         分别启停zk

                   ./zkServer.sh start

   ./zkServer.sh stop

                   ./zkServer.sh status 查看启动状态

         zk集群一般只有一个leader，多个follower，主一般是响应客户端的读写请求，而从主要同步数据，当主挂掉之后就会从follower里投票选举一个leader出来。

         可以用“jps”查看zk的进程.

五、kafka集群安装：

(1)下载软件

wget http://apache.opencas.org/kafka/0.9.0.1/kafka_2.11-0.9.0.1.tgz

(2)解压软件

tar -zxvf kafka_2.11-0.9.0.1.tgz     

(3)进入到config目录，主要修改server.config文件

 

以下为配置参数说明：

broker.id=0  #当前机器在集群中的唯一标识，和zookeeper的myid性质一样

port=9092 #当前kafka对外提供服务的端口默认是9092

host.name=192.168.7.100 #这个参数默认是关闭的，在0.8.1有个bug，DNS解析问题，失败率的问题。

num.network.threads=3 #这个是borker进行网络处理的线程数

num.io.threads=8 #这个是borker进行I/O处理的线程数

log.dirs=/opt/kafka/kafkalogs/ #消息存放的目录，这个目录可以配置为“，”逗号分割的表达式，上面的num.io.threads要大于这个目录的个数这个目录，如果配置多个目录，新创建的topic他把消息持久化的地方是，当前以逗号分割的目录中，那个分区数最少就放那一个

socket.send.buffer.bytes=102400 #发送缓冲区buffer大小，数据不是一下子就发送的，先回存储到缓冲区了到达一定的大小后在发送，能提高性能

socket.receive.buffer.bytes=102400 #kafka接收缓冲区大小，当数据到达一定大小后在序列化到磁盘

socket.request.max.bytes=104857600 #这个参数是向kafka请求消息或者向kafka发送消息的请请求的最大数，这个值不能超过java的堆栈大小

num.partitions=1 #默认的分区数，一个topic默认1个分区数

log.retention.hours=168 #默认消息的最大持久化时间，168小时，7天

message.max.byte=5242880  #消息保存的最大值5M

default.replication.factor=2  #kafka保存消息的副本数，如果一个副本失效了，另一个还可以继续提供服务

replica.fetch.max.bytes=5242880  #取消息的最大直接数

log.segment.bytes=1073741824 #这个参数是：因为kafka的消息是以追加的形式落地到文件，当超过这个值的时候，kafka会新起一个文件

log.retention.check.interval.ms=300000 #每隔300000毫秒去检查上面配置的log失效时间（log.retention.hours=168），到目录查看是否有过期的消息如果有，删除

log.cleaner.enable=false #是否启用log压缩，一般不用启用，启用的话可以提高性能

zookeeper.connect=192.168.177.5:2181,192.168.177.6:2181,192.168.177.7:2181#设置zookeeper的连接端口

(4)上面是参数的解释，实际的修改项为：

#broker.id=0  每台服务器的broker.id都不能相同
port=9092
#hostname
host.name=192.168.177.5
#在log.retention.hours=168 下面新增下面三项
message.max.byte=5242880
default.replication.factor=2
replica.fetch.max.bytes=5242880
#设置zookeeper的连接端口
zookeeper.connect=192.168.177.5:2181,192.168.177.6:2181,192.168.177.7:2181
 

(5)scp命令将配置好的kafka包文件拷贝到其它节点上

scp -r /it/ kafka_2.11-0.9.0.1/  root@192.168.177.6:/it/

修改其它节点上的参数，分别为：

broker.id=1，broker.id=2

host.name=192.168.177.6，host.name=192.168.177.7

port=9093，port=9094

六、启动Kafka集群并测试

(1)先启动3台节点的zookeeper服务
./zkServer.sh start
(2)启动Kafka(3台节点单独启动)
 bin/kafka-server-start.sh config/server.properties  >/dev/null  2>&1 &
(3)检查服务是否启动

(4)创建Topic

bin/kafka-topics.sh --create --zookeeper 192.168.177.5:2181 --replication-factor 3 --partitions 1--topic newtest

创建主题后，三台节点都能看到newtest主题创建成功。

(5)删除topic

bin/kafka-run-class.sh kafka.admin.TopicCommand --delete --topic demo --zookeeper 192.168.177.5:2181

(6)查看主题

bin/kafka-topics.sh --list --zookeeper 192.168.177.5:2181

(7)创建生产者producer

[root@lfl02kafka_2.10-0.8.2.0]# bin/kafka-console-producer.sh  --broker-list 192.168.177.5:9092  --topic  newtest

[2016-09-2019:28:27,544] WARN Property topic is not valid (kafka.utils.VerifiableProperties)

123

qwer

1234542345

hello

(8)创建消费着consumer

[root@lfl02kafka_2.10-0.8.2.0]# bin/kafka-console-consumer.sh  --zookeeper 192.168.177.5:2181  --topic newtest   --from-beginning

123

qwer

1234542345

Hello

消息可以成功消费到！

(9)此时杀掉server1上的broker1

pkill -9 -f config/server.properties

发现主题topic依然存活

[root@lfl02kafka_2.10-0.8.2.0]# bin/kafka-topics.sh --describe --zookeeper192.168.177.5:2181 --topic newtest

Topic:newtest PartitionCount:1      ReplicationFactor:3 Configs:

         Topic: newtest          Partition: 0       Leader:2 Replicas: 0,2,1          Isr: 2,1

(10)再次创建consumer，看看消息是否能正常消费到

[root@lfl02kafka_2.10-0.8.2.0]# bin/kafka-console-consumer.sh --zookeeper192.168.177.5:2181 --topic newtest --from-beginning

123

qwer

1234542345

Hello

说明一切正常！

 

七、kafka java调用：

(1)java端生产数据， kafka集群消费数据：

</pre><pre name="code" class="java">packagecom.sitech.dss.kafka.util;
importjava.util.Properties;
importjava.util.concurrent.TimeUnit;
importkafka.javaapi.producer.Producer;
importkafka.producer.KeyedMessage;
importkafka.producer.ProducerConfig;
importkafka.serializer.StringEncoder;
public class KafkaProducer extends Thread{
private String topic;
public KafkaProducer(String topic){
super();
this.topic = topic;
}
@Override
public void run() {
Producer producer = createProducer();
int i=0;
while(true){
producer.send(newKeyedMessage<Integer, String>(topic, "message: " + i++));
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e){
e.printStackTrace();
}
}
}
private Producer createProducer() {
Properties properties = new Properties();
properties.put("zookeeper.connect","192.168.177.5:2181,192.168.177.6:2181,192.168.177.7:2181");//声明zk
properties.put("serializer.class",StringEncoder.class.getName());
properties.put("metadata.broker.list","192.168.177.5:9092,192.168.177.6:9093,192.168.177.7:9094");// 声明kafka broker
return new Producer<Integer, String>(new ProducerConfig(properties));
}
public static void main(String[] args) {
new KafkaProducer("newtest").start();// 使用kafka集群中创建好的主题 newtest

}
}

(2)kafka集群中消费主题newtest的数据

[root@lfl02kafka_2.10-0.8.2.0]# bin/kafka-console-consumer.sh --zookeeper192.168.177.5:2181 --topic newtest --from-beginning

message:0

message:1

message:2

message:3

message:4

message:5

message:6

message:7

message:8

message:9

message:10

message:11

message:12

message:13

message:14

message:15

message:16

message:17

message:18

message:19

message:20

(3) kafka 使用Java写消费者，这样先运行kafkaProducer ，在运行kafkaConsumer，即可得到生产者的数据：

packagecom.sitech.dss.kafka.util;
importjava.util.HashMap;
importjava.util.List;
import java.util.Map;
importjava.util.Properties;
importkafka.consumer.Consumer;
importkafka.consumer.ConsumerConfig;
importkafka.consumer.ConsumerIterator;
importkafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
/**
* 接收数据
*
*/
public classKafkaConsumer extends Thread{

private String topic;
public KafkaConsumer(String topic){
super();
this.topic = topic;
}
@Override
public void run() {
ConsumerConnector consumer =createConsumer();
Map<String, Integer>topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, 1); // 一次从主题中获取一个数据
Map<String,List<KafkaStream<byte[], byte[]>>>  messageStreams = consumer.createMessageStreams(topicCountMap);
KafkaStream<byte[], byte[]>stream = messageStreams.get(topic).get(0);// 获取每次接收到的这个数据
ConsumerIterator<byte[], byte[]>iterator =  stream.iterator();
while(iterator.hasNext()){
String message = newString(iterator.next().message());
System.out.println("接收到: " + message);
}
}
private ConsumerConnector createConsumer(){
Properties properties = newProperties();
properties.put("zookeeper.connect","192.168.177.5:2181");//声明zk
properties.put("group.id","test-consumer-group1");// 必须要使用别的组名称， 如果生产者和消费者都在同一组，则不能访问同一组内的topic数据
properties.put("zookeeper.session.timeout.ms","40000");
properties.put("zookeeper.sync.time.ms", "200");
properties.put("auto.commit.interval.ms", "1000");;
returnConsumer.createJavaConsumerConnector(new ConsumerConfig(properties));
}
public static void main(String[] args){
newKafkaConsumer("newtest").start();// 使用kafka集群中创建好的主题 newtest

}

}

关于kafka说明可以参考：http://kafka.apache.org/documentation.html

OK，以上就是我对kafka的个人理解，不对之处请大家及时指正。