- 浏览: 663075 次
- 性别:
- 来自: 长沙
文章分类
- 全部博客 (364)
- quick start (57)
- bboss aop (43)
- bboss mvc (48)
- bboss persistent (96)
- bboss taglib (30)
- bboss event (10)
- bbossgroups (52)
- bboss (32)
- bboss会话共享 (17)
- bboss rpc (7)
- bboss 国际化 (5)
- bboss 序列化 (9)
- bboss cxf webservice (8)
- bboss hessian (3)
- bboss 安全认证SSO (15)
- bboss 工作流 (6)
- 平台 (18)
- bboss quartz (3)
- 杂谈 (5)
- 大数据 (1)
- bboss elastic (24)
- bboss http (1)
- bboss kafka (1)
- Elasticsearch Scroll和Slice Scroll查询API使用案例 (1)
最新评论
-
qianhao123:
...
采用gradle构建和发布bboss方法介绍 -
qianhao123:
[img][/img]
采用gradle构建和发布bboss方法介绍 -
yin_bp:
欢迎大家参与working
高性能elasticsearch ORM开发库使用介绍 -
qq641879434:
万分感谢
bboss 持久层sql xml配置文件编写和加载方法介绍 -
yin_bp:
qq641879434 写道怎么设置配置文件 可以查看执行的S ...
bboss 持久层sql xml配置文件编写和加载方法介绍
bboss kafka组件使用介绍
本文使用的实例对应的gradle源码工程git访问地址:
http://git.oschina.net/bboss/bestpractice
testkafka子工程地址
http://git.oschina.net/bboss/bestpractice/tree/master/testkafka
bboss kafka组件作用
1.导入bboss kafka组件
maven坐标
gradle坐标
其中kafka2x可以是1.1.0,2.3.0.2.7.0,2.8.0等,具体根据kafka服务端版本来选择
2.使用kafka producer,发送消息
2.1 kafka producer配置
编写kafka.xml配置文件,放到classpath跟路径下面
相关配置说明:
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消息
异步方式发送消息
<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根路径下面
配置说明:
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实现:
3.3 加载kafka consumer配置并启动消息接收线程
本文使用的实例对应的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-kafka2x</artifactId> <version>6.1.0</version> </dependency> 参考下面gradle补充kafka依赖包...
gradle坐标
api 'com.bbossgroups.plugins:bboss-plugin-kafka:6.1.0' api ( [group: 'org.apache.kafka', name: 'kafka_2.12', version: "${kafka2x}", transitive: true], ){ exclude group: 'log4j', module: 'log4j' exclude group: 'org.slf4j', module: 'slf4j-log4j12' } api ([group: 'org.apache.kafka', name: 'kafka-tools', version: "${kafka2x}", transitive: true],){ exclude group: 'log4j', module: 'log4j' exclude group: 'org.slf4j', module: 'slf4j-log4j12' exclude group: 'org.eclipse.jetty', module: 'jetty-server' exclude group: 'org.eclipse.jetty', module: 'jetty-servlets' exclude group: 'org.eclipse.jetty', module: 'jetty-servlet' exclude group: 'org.glassfish.jersey.containers', module: 'jersey-container-servlet' } api ([group: 'org.apache.kafka', name: 'kafka-clients', version: "${kafka2x}", transitive: true],){ exclude group: 'log4j', module: 'log4j' exclude group: 'org.slf4j', module: 'slf4j-log4j12' } api ([group: 'org.apache.kafka', name: 'kafka-streams', version: "${kafka2x}", transitive: true],){ exclude group: 'log4j', module: 'log4j' exclude group: 'org.slf4j', module: 'slf4j-log4j12' }
其中kafka2x可以是1.1.0,2.3.0.2.7.0,2.8.0等,具体根据kafka服务端版本来选择
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();
发表评论
-
bboss ioc快速入门教程
2017-08-13 11:28 1180bboss是一个非常不错的ioc框架,功能类似于spr ... -
bboss log4j滚动日志文件扩展插件使用介绍
2017-06-25 11:05 1364bboss扩展了log4j滚动切割文件插件org.apache ... -
bboss redis组件使用实例
2017-03-04 16:57 1089在工程中导入bboss redis组件 gradle comp ... -
bboss 与ecipse gradle buildship插件结合使用方法
2017-02-23 09:48 1132本文介绍bboss 与ecipse gradle builds ... -
如何快速高效地开发和调试基于gradle管理的web应用
2016-09-28 23:39 1379本文探讨如何高效快速地开发和调试基于gradle管理的web应 ... -
bboss wordpdf构建部署介绍
2016-09-02 15:47 507bboss wordpdf构建部署介绍 下载 源码下载地址: ... -
bboss gradle工程导入eclipse介绍
2016-07-24 14:01 2811bboss gradle工程导入eclipse介绍(本文适用于 ... -
采用gradle构建和发布bboss方法介绍
2016-05-01 23:23 4411采用gradle构建和发布bboss版本及从maven中央库下 ... -
bboss文件缓存组件FileContentCache介绍
2016-02-27 15:23 621bboss文件缓存组件FileContentCache介绍 ... -
bboss自动代码生成工具使用指南
2015-11-15 21:09 9937本文介绍bboss自动代码生成工具使用方法 工具在线试用: ... -
bboss应用程序运行容器使用介绍
2015-06-22 16:15 1840bboss微服务运行容器使 ... -
bboss框架配置监控介绍
2015-01-12 14:26 1568bboss框架配置监控介绍 ... -
bboss开发、模块工程目录结构及功能说明
2014-10-15 19:46 4693基于bboss开发项目说明 ... -
bboss最佳实践gradle工程清单及其作用介绍
2014-09-27 09:13 2135基于bboss开发项目说明 ... -
bboss版activiti 5.12扩展动态
2014-09-20 18:19 5153继扩展Activiti-5.12轻松实 ... -
bboss跨站攻击白名单和脚本攻击防火墙配置
2014-08-01 09:56 1013本文详细介绍bboss跨站攻击白名单和跨站脚本攻击防火墙配置 ... -
bboss会话共享demo使用指南
2014-07-01 17:57 5403为了方便应用系统集成b ... -
bboss序列化cglib代理对象方法介绍
2014-06-22 15:14 1075本文介绍bboss序列化cglib代理对象方法。经过cglib ... -
bboss session共享架构及特点
2014-06-14 12:33 6240bboss会话共享架构 bboss session共 ... -
bboss自定义类对象序列化机制介绍
2014-05-26 23:32 1368bboss自定义类对象序列 ...
相关推荐
bboss es特点请访问: https://www.oschina.net/p/bboss-elastic
http负载均衡组件-HttpRequestProxy 使用参考文档 负载均衡组件特点: 1.服务负载均衡(目前提供RoundRobin负载算法) 2.服务健康检查 3.服务容灾故障恢复 4.服务自动发现(zk,etcd,consul,eureka,db,其他第三...
NULL 博文链接:https://yin-bp.iteye.com/blog/2174863
bboss+es基本操作示例,elasticsearch,bboss,java使用demo
Elasticsearch rest client bboss介绍-Elastic2018中国开发者大会演讲稿
bboss会话共享培训文档,学习bboss不可多得的资料,值得拥有。
NULL 博文链接:https://yin-bp.iteye.com/blog/2325602
NULL 博文链接:https://yin-bp.iteye.com/blog/1449900
企业级 J2EE 开源框架:bboss
NULL 博文链接:https://yin-bp.iteye.com/blog/926166
springboot集成ElasticsearchBboss调用Elasticsearch的案例分享
资源目录: 项目源码(eshelloword.rar) bboss elasticsearch开发入门教程.wmv 搭建bboss elasticsearch开发环境.wmv 说明.txt
官方版本,亲测可用
NULL 博文链接:https://yin-bp.iteye.com/blog/2177475
浙江移动BBOSS系统快速入门手册V0.1归类.pdf
bboss-datatran 由 bboss 开源的数据采集&流批一体化工具,提供数据采集、数据清洗转换处理和数据入库以及数据指标统计计算流批一体化处理功能。
NULL 博文链接:https://yin-bp.iteye.com/blog/1131637
弹性Tran老板数据交换模块使用文档: : Bboss是一个很好的Elasticsearch Java Rest客户端。 它运行并访问像mybatis这样的elasticsearch来关联数据库。环境要求JDK要求:JDK 1.7+ Elasticsearch版本要求:1.X,2.X,5...
基于bboss新版平台项目环境搭建和代码工具使用视频教程
官方版本,亲测可用