开发者社区> bboss> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

bboss kafka组件使用介绍

简介: bboss kafka组件使用介绍 本文使用的实例对应的gradle源码工程git访问地址: http://git.oschina.net/bboss/bestpractice testkafka子工程地址 http://git.
+关注继续查看
bboss kafka组件使用介绍
本文使用的实例对应的gradle源码工程git访问地址:
http://git.oschina.net/bboss/bestpractice
testkafka子工程地址
http://git.oschina.net/bboss/bestpractice/tree/master/testkafka
bboss kafka组件作用
  • 快速配置kafka客户端和消费者
  • 发送数据到kafka
  • 从kafka接收和处理数据(支持批量消息处理和按条处理)

1.导入bboss kafka组件
maven坐标
<dependency>
    <groupId>com.bbossgroups.plugins</groupId>
    <artifactId>bboss-plugin-kafka</artifactId>
    <version>5.0.6.2</version>
</dependency>

gradle坐标
compile 'com.bbossgroups.plugins:bboss-plugin-kafka:5.0.6.2'



2.使用kafka producer,发送消息
2.1 kafka producer配置
编写kafka.xml配置文件,放到classpath跟路径下面
<properties>
	<property name="productorPropes">
		<propes>
			
			<property name="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer">
				<description> <![CDATA[ 指定序列化处理类,默认为kafka.serializer.DefaultEncoder,即byte[] ]]></description>
			</property>
			<property name="key.serializer" value="org.apache.kafka.common.serialization.LongSerializer">
				<description> <![CDATA[ 指定序列化处理类,默认为kafka.serializer.DefaultEncoder,即byte[] ]]></description>
			</property>
					
			<property name="compression.type" value="gzip">
				<description> <![CDATA[ 是否压缩,默认0表示不压缩,1表示用gzip压缩,2表示用snappy压缩。压缩后消息中会有头来指明消息压缩类型,故在消费者端消息解压是透明的无需指定]]></description>
			</property>
			<property name="bootstrap.servers" value="hadoop85:9092,hadoop86:9092,hadoop88:9092">
				<description> <![CDATA[ 指定kafka节点列表,用于获取metadata(元数据),不必全部指定]]></description>
			</property>
		</propes>
	</property>
        <property name="workerThreadSize" value="100"/>
        <property name="workerThreadQueueSize" value="10240"/>
	<property name="kafkaproductor" 
		class="org.frameworkset.plugin.kafka.KafkaProductor"
		init-method="init"
		f:sendDatatoKafka="true"
f:sendAsyn="true"
		f:productorPropes="attr:productorPropes"/>		 
		
</properties>


相关配置说明:

bootstrap.servers kafka服务器地址配置
value.serializer kafka消息序列化插件配置
key.serializer kafka消息key序列化插件配置
f:sendDatatoKafka="true" 是否启动消息发送功能,false 禁用,true 启用
f:sendAsyn="true" 控制组件是否异步发送消息,默认为true
workerThreadSize 异步发送消息线程池,默认100
workerThreadQueueSize 异步发送消息队列,默认10240

2.2 发送kafka消息

发送kafka消息相关组件:
org.frameworkset.plugin.kafka.KafkaUtil
org.frameworkset.plugin.kafka.KafkaProductor

KafkaUtil组件加载配置文件并获取KafkaProductor ,通过KafkaProductor 发送kafka消息
KafkaProductor productor = KafkaUtil.getKafkaProductor("kafkaproductor");
		productor.send("blackcat",//kafka topic
				1l, //message key
				"aaa");//message
		productor.send("blackcat", //kafka topic
				"bbb"); //message


异步方式发送消息

<property name="workerThreadSize" value="100"/>
<property name="workerThreadQueueSize" value="10240"/>

<property name="kafkaproductor"
class="org.frameworkset.plugin.kafka.KafkaProductor"
init-method="init"
f:sendDatatoKafka="true"
f:sendAsyn="true"
f:productorPropes="attr:productorPropes"/>

通过api控制是否异步发送消息:

//异步方式发送消息
productor.send("blackcat",3l,"aaa",true);
productor.send("blackcat",4l,"bbb",true);

//同步方式发送消息
productor.send("blackcat",5l,"aaa",false);
productor.send("blackcat",6l,"bbb",false);

3.接收和处理kafka消息
3.1 kafka consumer配置
新建kafkaconsumer.xml文件,放到classpath根路径下面
<properties>
	<property name="consumerPropes">
		<propes>


			<property name="group.id" value="test">
				<description> <![CDATA[ 指定kafka group id]]></description>
			</property>
			<property name="zookeeper.session.timeout.ms" value="30000">
				<description> <![CDATA[ 指定kafkazk会话超时时间]]></description>
			</property>
			

			<property name="auto.commit.interval.ms" value="3000">
				<description> <![CDATA[ 指定kafka自动提交时间间隔]]></description>
			</property>

			<property name="auto.offset.reset" value="smallest">
				<description> <![CDATA[ ]]></description>
			</property>
			<property name="zookeeper.connect" value="hadoop85:2181,hadoop86:2181,hadoop88:2181">
				<description> <![CDATA[ 指定kafka节点列表,用于获取metadata(元数据),不必全部指定]]></description>
			</property>

		</propes>
	</property>
	<property name="kafkaconsumer"
		class="org.frameworkset.plugin.kafka.KafkaBatchConsumer" init-method="init"
f:batchsize="-1"
		f:checkinterval="10000"
		f:productorPropes="attr:consumerPropes" f:topic="blackcat"
		f:storeService="attr:storeService" f:partitions="4" />
	<property name="storeService"
		 class="org.frameworkset.plugin.kafka.StoreServiceTest" />	

</properties>

配置说明:
storeService 配置消息处理组件
zookeeper.connect 配置管理kafka服务器和消息的zookeeper集群地址
f:topic="blackcat" 消费的kafka topic
f:partitions="4" topic对应的分区数,决定并行处理消息的工作线程
f:batchsize="-1" 批处理消息条数,-1禁用批处理,>0时按照批处理方式按批次提交消息给storeservice组件
f:checkinterval="10000" 指定批处理消息接收最大等待时间,单位毫秒。按照批处理方式时,如果超过checkinterval指定的时间,到达的消息没有到达batchsize,则强制提交处理当前批次的数据到storeservice组件

3.2 接收和处理消息
接收和处理消息相关组件:
org.frameworkset.plugin.kafka.KafkaConsumer
org.frameworkset.plugin.kafka.StoreService

编写消息处理组件,处理组件需要实现接口
org.frameworkset.plugin.kafka.StoreService
//按条处理数据
public void store(MessageAndMetadata<byte[], byte[]> message)  throws Exception ;
public void closeService();
//按批处理消息
public void store(List<MessageAndMetadata<byte[], byte[]>> messages) throws Exception
StoreServiceTest实现:
package org.frameworkset.plugin.kafka;

import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import kafka.message.MessageAndMetadata;

public class StoreServiceTest extends BaseStoreService {
	StringDeserializer sd = new StringDeserializer();
	LongDeserializer ld = new LongDeserializer();
	@Override
	public void store(List<MessageAndMetadata<byte[], byte[]>> messages) throws Exception {
		for(MessageAndMetadata<byte[], byte[]> message:messages){
			String data = sd.deserialize(null,message.message());
			long key = ld.deserialize(null, message.key());
			System.out.println("key="+key+",data="+data);
		}
	}

	@Override
	public void closeService() {
		sd.close();
		ld.close();
	}

	@Override
	public void store(MessageAndMetadata<byte[], byte[]> message) throws Exception {
		String data = sd.deserialize(null,message.message());
		long key = ld.deserialize(null, message.key());
		System.out.println("key="+key+",data="+data);
	}

}


3.3 加载kafka consumer配置并启动消息接收线程
BaseApplicationContext context = DefaultApplicationContext.getApplicationContext("kafkaconfumer.xml");
		KafkaListener consumer = context.getTBeanObject("kafkaconsumer", KafkaListener.class);
		Thread t = new Thread(consumer);
		t.start();



版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
MacOs安装Kafka
一、安装kafka brew install kafka
103 0
bboss ioc配置文件中使用外部属性文件介绍
bboss ioc配置文件中使用外部属性文件介绍 与spring ioc一样,在bboss ioc中也可以非常方便地引用外部属性文件(5.0.1及后续版本),本文介绍使用方法。 在工程中引入bboss ioc: maven坐标: com.
720 0
bboss分布式事件框架介绍文档
bboss分布式事件框架介绍文档,文档中介绍了bboss分布式事件框架产生的背景、架构、应用集成配置以及开发API。
688 0
bboss开发、模块工程目录结构及功能说明
基于bboss开发项目说明 要做简单的demo,请参考文档 http://yin-bp.iteye.com/blog/1026261 正儿八经的做项目,参考文档搭bboss平台开发环境: http://yin-bp.
1145 0
[Mac OS/iOS]反汇编工具Hopper分析Crash Log
   在Mac OS下分析Crash Log有很多种方法,这里不是要说明如何分析的Crash Log, 主要是展示下Hopper的使用。 强大的IDA大家可能已经知道,但它的Mac OS版本又让人回到了DOS时代。
2032 0
bboss mvc快速入门教程
基于bboss开发项目说明 要做简单的demo,请参考文档 http://yin-bp.iteye.com/blog/1026261 正儿八经的做项目,参考文档搭bboss平台开发环境 普通ui版平台: http://yin-bp.
876 0
bboss aop 远程服务介绍-网络环境
bboss 项目下载地址: https://sourceforge.net/project/showfiles.php?group_id=238653 网络的要求就是要求bboss aop中的远程管理组件能够在各个服务器之间进行通讯,根据采用的不同网络协议,说明如下: Udp协议:确保能够进行udp通讯,如果你的机器之间装有防火墙之类的软件,必须允许udp数据包的传递。
687 0
+关注
bboss
热衷开源,主要开源作品:开源框架bboss,bboss session,bboss 序列化等 bboss特色:http://yin-bp.iteye.com/blog/1080824
328
文章
1
问答
文章排行榜
最热
最新