kafka producer实例及原理分析-阿里云开发者社区

1.前言

首先，描述下应用场景：

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

步骤如下：

搭建KAFKA系统运行环境

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

http://zhangfengzhe.blog.51cto.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<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