1.配置文件 consumer.properties
#zookeeper地址
zookeeper.connect=master:2181,slave1:2181,slave2:2181
#zookeeper超时时间
zookeeper.connectiontimeout.ms=1000000
#kafka的consumer组
group.id=test-group
2. 组织代码
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
public class init {
public static void main(String[] args) throws FileNotFoundException, IOException {
//1、读取配置文件
Properties p = new Properties();
//p.load(new FileInputStream(new File("./consumer.properties")));
p.load(init.class.getClassLoader().getResourceAsStream("consumer.properties"));
//2、通过配置文件,创建消费者配置
ConsumerConfig config = new ConsumerConfig(p);
//3、通过配置,创建消费者
ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);
//4、创建封装消息的map
Map<String,Integer> topics = new HashMap<String,Integer>();
//5、封装消费的topic名字和消费这个topic中几个partition
topics.put("test-topic", 1);
//6、创建消息流,传入刚刚创建的map
Map<String, List<KafkaStream<byte[], byte[]>>> streams = consumer.createMessageStreams(topics);
//7、获取消息,封装到list中
List<KafkaStream<byte[], byte[]>> partitions = streams.get("test-topic");
//8、创建线程池,用不同的线程消费list中的不同消息,可以消费一个,也可以消费多个
ExecutorService threadPool = Executors.newFixedThreadPool(1);
for(KafkaStream<byte[], byte[]> partition : partitions){
//9、将消息传到线程中消费
threadPool.execute(new TestConsumer(partition));
}
}
}
3.调用显示代码
import java.io.UnsupportedEncodingException;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.message.MessageAndMetadata;
public class TestConsumer extends Thread {
private KafkaStream<byte[], byte[]> partition;
/**
* 构造函数,传进来消息
*/
public TestConsumer(KafkaStream<byte[], byte[]> partition){
this.partition = partition;
}
public void run() {
//1、取出消息
ConsumerIterator<byte[], byte[]> iterator = partition.iterator();
//2、判断是否有下一条,有的话迭代消息,没有的话,停止等待
while(iterator.hasNext()){
//3、取出消息
MessageAndMetadata<byte[], byte[]> next = iterator.next();
try {
//4、打印消息
System.out.println("partiton:" + next.partition());//partition
System.out.println("offset:" + next.offset()); //偏移量
System.out.println("message:" + new String(next.message(), "utf-8"));//消息主体
} catch (UnsupportedEncodingException e){
e.printStackTrace();
}
}
}
分享到:
相关推荐
kafka生产者消费者实例
kafka大数据 生产者消费者实例,java代码实现,完整版
kafka实例 消费者生产者,实现代码,可使用maven 下载包pom.xml
Kafka 是一种高吞吐量 的分布式发布订阅消息系统,本实例利用C#开发,同步数据,亲测可用!
主要介绍了Java实现Kafka生产者消费者代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
kafka模拟生产者、消费者,集群模式,若是单机版,将ip端口组改为相应ip端口即可;
kafka学习过程,maven工程,包含基础过程、提升过程。可供大家学习一下,里面有详细注释,一个groupid多个Consumer来消费消息和一个Consumer且有多个线程消费
kafka集群测试,生产、消费实例代码,kafka-demo.rar
示例Golang Kafka消费者和生产者 设置 制片人 go run cmd/producer/main.go 消费者 标志: brokerList 话题 分区(默认值:0) offsetType messageCountStart(默认值:0) go run cmd/consumer/main.go ...
简单的Kafka消费者-生产者示例运行项目的步骤下载Kafka 0.10.0.0二进制文件cd kafka_2.11-0.10.0.0 运行zookeeper和kafka服务器bin/zookeeper-server-start.sh config/zookeeper.properties bin/kafka-server-start....
Kafka基础实战:消费者和生产者实例
from kafka import KafkaConsumer, TopicPartition, OffsetAndMetadata from consumers.db_util import * from consumers.json_dispose import * from collections import OrderedDict threads = []
kafka生产者和消费者实例,了解Kafka的一个简单入门实例源码下载
kafka消息队列消费者生产者读写样例。Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。
卡夫卡消费者的破坏者演示如何在Kafka 0.9 Consumer上使用LMAX Disruptor 好处->一旦先前的使用者完全处理完消息,便可以使用序列屏障来提交消息。想象力是极限。如果环形缓冲区可以容纳在L3缓存中,则处理速度会更...
Kafka是一种高吞吐量的分布式发布订阅消息系统,它可以处理消费者规模的网站中的所有动作流数据。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理,也是为了通过集群机来提供实时的消费。文档介绍...
卡夫卡消费者经理 该软件包用于简化kafka消费者的常用用法: 通过实例创建支持多个kafka-consumer-manager 提供对autoCommit的支持:通过将消息保存到队列中并按每个分区的消息保存工作消息来进行错误和限制(并发...
简单介绍了kafka的入门操作,快速搭建一个实例,以及讲解了kafka消费者组的概念
kafkaDemo,通过API向指定主题生产数据,创建topic代码,删除topic代码,创建消费者线程指定消费者组
消费 者 和生产者,代码为 idea 上的 maven项目,如有疑问请留言