Kafka Producer端封装自定义消息

原创
2013/01/07 18:30
阅读数 1.8W

这篇文章主要讲kafka producer端的编程,通过一个应用案例来描述kafka在实际应用中的作用。如果你还没有搭建起kafka的开发环境,可以先参考:<kafka开发环境搭建>

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

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

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

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

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

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

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

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

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

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

@Override
	public String toString() {
		String keyword = "[info kafka producer:]";
		keyword = keyword + this.getId() + "-" + this.getUser() + "-"
				+ this.getKeyword() + "-" + this.getCurrent();
		return keyword;
	}

这样还没有完成,这只是将数据格式用java对象表现出来,解析来要对其按照kafka的消息类型进行封装,在这里我们只需要实现Encoder类即可:

package org.gfg.kafka.message;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.message.Message;

public class KeywordMessage implements kafka.serializer.Encoder<Keyword>{
	
	public static final Logger LOG=LoggerFactory.getLogger(Keyword.class); 
	
	@Override
	public Message toMessage(Keyword words) {
		LOG.info("start in encoding...");
		return new Message(words.toString().getBytes());
	}

}

注意泛型和返回类型即可。这样KeywordMessage就是一个可以被kafka发送和存储的对象了。

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

/**配置producer必要的参数*/
		Properties props = new Properties();
		props.put("zk.connect", "192.168.10.11:2181");
		/**选择用哪个类来进行序列化*/
		props.put("serializer.class", "org.gfg.kafka.message.KeywordMessage");
		props.put("zk.connectiontimeout.ms", "6000");
		ProducerConfig config=new ProducerConfig(props);
		
		/**制造数据*/
		Keyword keyword=new Keyword();
		keyword.setUser("Chenhui");
		keyword.setId(0);
		keyword.setKeyword("china");
		
		List<Keyword> msg=new ArrayList<Keyword>();
		msg.add(keyword);
		
		/**构造数据发送对象*/
		Producer<String, Keyword> producer=new Producer<String, Keyword>(config);		
		ProducerData<String,Keyword> data=new ProducerData<String, Keyword>("test", msg);
		producer.send(data);

发送完之后,我们可以用bin目录下的kafka-console-consumer来看发送的结果(当然现在用的topictest)。可以用命令:

./kafka-console-consumer –zookeeper 192.168.10.11:2181 –topic test –from-beginning

如果是在使用zookeeper搭建分布式的情况下(zookeeper based broker discovery),我们可以执行第三个步骤,用编码来实现partition的分配策略。这里需要我们实现Partitioner对象:

package org.gfg.kafka.partitioner;

import org.gfg.kafka.message.Keyword;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import kafka.producer.Partitioner;

/**
 * 
 * @author Chen.Hui
 * 
 */
public class ProducerPartitioner implements Partitioner<String> {
	
	public static final Logger LOG=LoggerFactory.getLogger(Keyword.class); 
	
	@Override
	public int partition(String key, int numPartitions) {
		LOG.info("ProducerPartitioner key:"+key+" partitions:"+numPartitions);
		return key.length() % numPartitions;
	}

}

在上面的partition方法中,值得注意的是,key我们是在构造数据发送对象时设置的,这个key是区分存储的关键,比如我想将我的数据按照不同的用户类别存储。Partition的好处是可以并发的获取同类数据,提高效率,具体可以看之前的文章。

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

/**选择用哪个类来进行设置partition*/
		props.put("partitioner.class", "org.gfg.kafka.partitioner.ProducerPartitioner");

		ProducerData<String,Keyword> data=new ProducerData<String, Keyword>("test","developer", msg);

增加了对partition的配置,并且修改了ProducerData的参数,其中,中间的就是key,如果不设置partitionkafka则随机的向broker中发送请求。我们可以看一眼ProducerData的源码:

package kafka.javaapi.producer

import scala.collection.JavaConversions._

class ProducerData[K, V](private val topic: String,
                         private val key: K,
                         private val data: java.util.List[V]) {

  def this(t: String, d: java.util.List[V]) = this(topic = t, key = null.asInstanceOf[K], data = d)

  def this(t: String, d: V) = this(topic = t, key = null.asInstanceOf[K], data = asList(List(d)))

  def getTopic: String = topic

  def getKey: K = key

  def getData: java.util.List[V] = data
}

至此,producer端的事情都做完了,当然这就是个demo,还有很多性能上的优化需要做,当然有了这个基础,我们就能将数据存储到broker上,下一步,就是用consumer来消费这些日志,形成有价值的数据产品。

展开阅读全文
打赏
3
24 收藏
分享
加载中
这里producer当前0.8.2版本已经没有zk.connect配置了,只有metadata.broker.list,这样好像没办法做高可用,一个broker节点宕掉,发送消息就会报错
2015/10/06 21:07
回复
举报
请问楼主 ,用的是kafka哪个版本, 初学,感觉kafka的api个版本这件不太兼容啊
2015/04/03 11:11
回复
举报
淘宝数据服务平台的是否有一个consumer将数据持久化到HDFS或云存储上?
2014/06/26 14:42
回复
举报
Gaischen博主

引用来自“yang009ww”的评论

Producer端可以给Consumer端发消息,反之,Consumer端能给Producer端发消息吗(根据你前面的文章讲的是Producer是push的动作,而Consumer是pull的动作,这样的话是不是指Consumer端无法向Producer端push消息呢?因为Consumer不能push,Producer不能pull)?如果kafka运用到项目中,比如java语言编写Consumer端,如何能够实现在Linux环境下Producer端发送新消息时,Consumer端能够自动接收?

这个和socket是不一样的 虽然消息系统可以做通信用,但是和一般的socket写的通信不太一样,这个主要作用还是高吞吐量的处理、存储日志。
consumer的用法你可以看后面的文章,有写到的,有你说的这种用法
2013/03/08 16:05
回复
举报
Producer端可以给Consumer端发消息,反之,Consumer端能给Producer端发消息吗(根据你前面的文章讲的是Producer是push的动作,而Consumer是pull的动作,这样的话是不是指Consumer端无法向Producer端push消息呢?因为Consumer不能push,Producer不能pull)?如果kafka运用到项目中,比如java语言编写Consumer端,如何能够实现在Linux环境下Producer端发送新消息时,Consumer端能够自动接收?
2013/03/08 15:48
回复
举报
更多评论
打赏
5 评论
24 收藏
3
分享
返回顶部
顶部