您的位置：首页 > 其它

Kafka Producer端自定义消息

2014-03-12 22:45 253 查看
这篇文章主要讲kafkaproducer端的编程，通过一个应用案例来描述kafka在实际应用中的作用。如果你还没有搭建起kafka的开发环境，可以先参考：<kafka开发环境搭建>

首先描述一下应用的情况：一个站内的搜索引擎，运营人员想知道某一时段，各类用户对商品的不同需求。通过对这些数据的分析，从而获得更多有价值的市场分析报表。这样的情况，就需要我们对每次的搜索进行记录，当然，不太可能使用数据库区记录这些信息（数据库存这些数据我会觉得是种浪费，个人意见）。最好的办法是存日志。然后通过对日志的分析，计算出有用的数据。我们采用kafka这种分布式日志系统来实现这一过程。

完成上述一系列的工作，可以按照以下步骤来执行：

1.搭建kafka系统运行环境。

2.设计数据存储格式（按照自定义格式来封装消息）

3.Producer端获取真实数据（搜索记录），并对数据按上述2中设计的格式进行编码。

4.Producer将已经编码的数据发送到broker上，在broker上进行存储（分配存储策略）。

5.Consumer端从broker中获取数据，分析计算。

如果用淘宝数据服务平台的架构来匹配这一过程，broker就好比数据中心中存储的角色，producer端基本是放在了应用中心的开放API中，consumer端则一般用于数据产品和应用中心的获取数据中使用。





今天主要写的是2、3、4三个步骤。我们先看第二步。为了快速实现，这里就设计一个比较简单的消息格式，复杂的原理和这个一样。




用四个字段分别表示消息的ID、用户、查询关键词和查询时间。当然你如果要设计的更复杂，可以加入IP这些信息。这些用java写就是一个简单的pojo类，这是getter/setter方法即可。由于在封转成kafka的message时需要将数据转化成bytep[]类型，可以提供一个序列化的方法。我在这里直接重写toString了：

1
@Override
2
public
String
toString(){
3
String
keyword=
"[info
kafkaproducer:]"
;
4
keyword
=keyword+
this
.getId()
+
"-"
+
this
.getUser()
+
"-"
5
+
this
.getKeyword()
+
"-"
+
this
.getCurrent();
6
return
keyword;
7
}
这样还没有完成，这只是将数据格式用java对象表现出来，解析来要对其按照kafka的消息类型进行封装，在这里我们只需要实现Encoder类即可：

01
package
org.gfg.kafka.message;
02
03
import
org.slf4j.Logger;
04
import
org.slf4j.LoggerFactory;
05
06
import
kafka.message.Message;
07
08
public
class
KeywordMessage
implements
kafka.serializer.Encoder<Keyword>{
09
10
public
static
final
Logger
LOG=LoggerFactory.getLogger(Keyword.
class
);
11
12
@Override
13
public
Message
toMessage(Keywordwords){
14
LOG.info(
"start
inencoding..."
);
15
return
new
Message(words.toString().getBytes());
16
}
17
18
}
注意泛型和返回类型即可。这样KeywordMessage就是一个可以被kafka发送和存储的对象了。

接下来，我们可以编写一部分producer，获取业务系统的数据。要注意，producer数据的推送到broker的，所以发起者还是业务系统，下面的代码就能直接发送一次数据，注释都很详细：

01
/**配置producer必要的参数*/
02
Properties
props=
new
Properties();
03
props.put(
"zk.connect"
,
"192.168.10.11:2181"
);
04
/**选择用哪个类来进行序列化*/
05
props.put(
"serializer.class"
,
"org.gfg.kafka.message.KeywordMessage"
);
06
props.put(
"zk.connectiontimeout.ms"
,
"6000"
);
07
ProducerConfig
config=
new
ProducerConfig(props);
08
09
/**制造数据*/
10
Keyword
keyword=
new
Keyword();
11
keyword.setUser(
"Chenhui"
);
12
keyword.setId(
0
);
13
keyword.setKeyword(
"china"
);
14
15
List<Keyword>
msg=
new
ArrayList<Keyword>();
16
msg.add(keyword);
17
18
/**构造数据发送对象*/
19
Producer<String,
Keyword>producer=
new
Producer<String,
Keyword>(config);
20
ProducerData<String,Keyword>
data=
new
ProducerData<String,
Keyword>(
"test"
,
msg);
21
producer.send(data);
发送完之后，我们可以用bin目录下的kafka-console-consumer来看发送的结果（当然现在用的topic是test）。可以用命令：

1
./kafka-console-consumer
–zookeeper192.168.10.11:2181–topic
test
–from-beginning




如果是在使用zookeeper搭建分布式的情况下（zookeeperbasedbrokerdiscovery），我们可以执行第三个步骤，用编码来实现partition的分配策略。这里需要我们实现Partitioner对象：

01
package
org.gfg.kafka.partitioner;
02
03
import
org.gfg.kafka.message.Keyword;
04
import
org.slf4j.Logger;
05
import
org.slf4j.LoggerFactory;
06
07
import
kafka.producer.Partitioner;
08
09
/**
10
*
11
*
@authorChen.Hui
12
*
13
*/
14
public
class
ProducerPartitioner
implements
Partitioner<String>
{
15
16
public
static
final
Logger
LOG=LoggerFactory.getLogger(Keyword.
class
);
17
18
@Override
19
public
int
partition(String
key,
int
numPartitions)
{
20
LOG.info(
"ProducerPartitioner
key:"
+key+
"
partitions:"
+numPartitions);
21
return
key.length()
%numPartitions;
22
}
23
24
}
在上面的partition方法中，值得注意的是，key我们是在构造数据发送对象时设置的，这个key是区分存储的关键，比如我想将我的数据按照不同的用户类别存储。Partition的好处是可以并发的获取同类数据，提高效率，具体可以看之前的文章。

所以在第二部时的producer代码需要有所改进：

1
/**选择用哪个类来进行设置partition*/
2
props.put(
"partitioner.class"
,
"org.gfg.kafka.partitioner.ProducerPartitioner"
);
3
4
ProducerData<String,Keyword>
data=
new
ProducerData<String,
Keyword>(
"test"
,
"developer"
,
msg);
增加了对partition的配置，并且修改了ProducerData的参数，其中，中间的就是key，如果不设置partition，kafka则随机的向broker中发送请求。我们可以看一眼ProducerData的源码：

01
package
kafka.javaapi.producer
02
03
import
scala.collection.JavaConversions.
_
04
05
class
ProducerData[K,
V](
private
val
topic
:
String,
06
private
val
key
:
K,
07
private
val
data
:
java.util.List[V])
{
08
09
def
this
(t
:
String,
d
:
java.util.List[V])
=
this
(topic
=
t,
key
=
null
.asInstanceOf[K],
data
=
d)
10
11
def
this
(t
:
String,
d
:
V)
=
this
(topic
=
t,
key
=
null
.asInstanceOf[K],
data
=
asList(List(d)))
12
13
def
getTopic
:
String
=
topic
14
15
def
getKey
:
K
=
key
16
17
def
getData
:
java.util.List[V]
=
data
18
}
至此，producer端的事情都做完了，当然这就是个demo，还有很多性能上的优化需要做，当然有了这个基础，我们就能将数据存储到broker上，下一步，就是用consumer来消费这些日志，形成有价值的数据产品
                                            

                                        

                        
                        内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 点击这里给我发消息 
                    

                    

                    

                        

                         标签： 
                                                

                        

                    


                

            


            

                
相关文章推荐

                

                
            

        

        

            

            
            

                

                    
新的分享

                    

                        
                    

                

            


            

                

                    
章节导航