kafka producer实例及原理分析-白红宇

1.前言

首先，描述下应用场景：

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

步骤如下：

搭建KAFKA系统运行环境

如果你还没有搭建起来，可以参考我的博客：

设计数据存储格式

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<MobileGameLog>{ 
@Override
public
byte
[] toBytes(MobileGameLog mobileGameLog) { 
return
mobileGameLog.toString().getBytes();
}
public
MobileGameKafkaMessage(VerifiableProperties props){ 
}
}
            
          

Step 3 : 提供Partitioner

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

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

Step 4 : 提供Producer

提供配置

运行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<key , value>
// V: type of the message
// K: type of the optional key associated with the message
kafka.javaapi.producer.Producer<MobileGameLog, MobileGameLog> producer 
= 
new
Producer<MobileGameLog, MobileGameLog>(
config);
List<KeyedMessage<MobileGameLog, MobileGameLog>> list 
= 
new
ArrayList<KeyedMessage<MobileGameLog, MobileGameLog>>();
// 5条tlbb数据
for
(
int
i = 
1
; i <= 
5
; i++) { 
MobileGameLog log = 
new
MobileGameLog();
log.setActionType(
"YuanBaoShop"
);
log.setAppKey(
"tlbb"
);
log.setGuid(
"xxx_"
+ i);
log.setTime(
"2014-10-01 10:00:20"
);
KeyedMessage<MobileGameLog, MobileGameLog> keyedMessage 
= 
new
KeyedMessage<MobileGameLog, MobileGameLog>(
"log_1"
, log, log);
list.add(keyedMessage);
}
// 8条ldj数据
for
(
int
i = 
1
; i <= 
8
; i++) { 
MobileGameLog log = 
new
MobileGameLog();
log.setActionType(
"BlackMarket"
);
log.setAppKey(
"ldj"
);
log.setGuid(
"yyy_"
+ i);
log.setTime(
"2014-10-02 10:00:20"
);
KeyedMessage<MobileGameLog, MobileGameLog> keyedMessage 
= 
new
KeyedMessage<MobileGameLog, MobileGameLog>(
"log_1"
, log, log);
list.add(keyedMessage);
}
producer.send(list);
producer.close();
            
          

说明：

a.producer既可以send 一个keyedMessage,可以是一个keyedMessage list.

b.注意producer实例化时的泛型。value是消息对象，即POJO，key是这个pojo的标示，这个是要用来进行分区的。

c.producer向broker发送的是KeyedMessage,注意实例化时的泛型，KEY/VALUE的意义同b.

d.KeyedMessage需要指明topic name.

eclipse 运行结果如下：

-------start info

运行至MobileGameKafkaPartition

VerifiableProperties : {metadata.broker.list=192.168.152.2:9092,192.168.152.2:9093,

zk.connectiontimeout.ms=6000, request.required.acks=1,

partitioner.class=com.sohu.game.kafka.day2.MobileGameKafkaPartition,

serializer.class=com.sohu.game.kafka.day2.MobileGameKafkaMessage}

-------end info

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".

SLF4J: Defaulting to no-operation (NOP) logger implementation

SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.

-------start info