kafka producer实例及原理分析

QQ叫紫珊

　　1.前言
　　首先，描述下应用场景：

假设，公司有一款游戏，需要做行为统计分析，数据的源头来自日志，由于用户行为非常多，导致日志量非常大。将日志数据插入数据库然后再进行分析，已经满足不了。最好的办法是存日志，然后通过对日志的分析，计算出有用的数据。我们采用kafka这种分布式日志系统来实现这一过程。

　　步骤如下：

搭建KAFKA系统运行环境 　　 　　如果你还没有搭建起来，可以参考我的博客： 　　http://zhangfengzhe.blog.运维网.com/8855103/1556650 　　 　　设计数据存储格式 　　 　　 Producer端获取数据，并对数据按上述设计的格式进行编码 Producer将已经编码的数据发送到broker上，在broker上进行存储 　　 Consumer端从broker中获取数据，分析计算。

　　

　　2.实现过程

　　

　　为了快速实现，我们简化日志消息格式。
　　在eclipse新建JAVA PROJECT，将kafka/libs下*.jar配置到项目build path即可。
　　

　　Step 1 : 简单的POJO对象（MobileGameLog）

private String actionType; private String appKey; private String guid; private String time;　　 　　说明： 　　actionType 代表行为类型 　　appKey     代表游戏ID 　　guid       代表角色 　　time       代表时间 　　 　　提供getter/setter方法，并override toString()

　　

　　Step 2 : 提供serializer
　　

　　需要注意的是，POJO对象需要序列化转化成KAFKA识别的消息存储格式--byte[]

　　

public class MobileGameKafkaMessage implements kafka.serializer.Encoder{ @Override public byte[] toBytes(MobileGameLog mobileGameLog) { return mobileGameLog.toString().getBytes(); } public MobileGameKafkaMessage(VerifiableProperties props){ } }

　　

　　

　　Step 3 : 提供Partitioner
　　

　　我们可以提供Partitioner，这样可以使得数据按照我们的策略来存储在brokers中。
　　

　　这里，我根据appKey来进行分区。
　　

　　

　　Step 4 : 提供Producer
　　

[table][tr][td=1,1,747]　　

• 　　提供配置
  

　　

　　

• 　　运行kafka环境
  

　　
启动zookeeper:
[root@localhost kafka_2.9.2-0.8.1.1]# bin/zookeeper-server-start.sh  
config/zookeeper.properties &　　

　　启动kafka broker(id=0):
[root@localhost kafka_2.9.2-0.8.1.1]# bin/kafka-server-start.sh
config/server.properties &　　

　　启动kafka broker(id=1)
[root@localhost kafka_2.9.2-0.8.1.1]# bin/kafka-server-start.sh  
config/server-1.properties &　　

　　上述过程，在我的博客【搭建kafka运行环境】里面都有详细记录，大家可以参考。
　　

　　

　　创建一个topic:
[root@localhost kafka_2.9.2-0.8.1.1]# bin/kafka-topics.sh --zookeeper localhost:2181
--create --topic log_1 --replication-factor 2 --partitions 3　　注意topic:log_1有3个分区，2个复制。
　　

　　

　　

• 　　制造数据并发送
  

　　

// Producer
// V: type of the message
// K: type of the optional key associated with the message
kafka.javaapi.producer.Producer producer
= new Producer(
config);
List list
= new ArrayList();
// 5条tlbb数据
for (int i = 1; i