介绍apache kafka 是分布式发布-订阅消息系统，在 kafka官网上对 kafka 的定义：一个分布式发布-订阅消息传递系统。 它最初由linkedin公司开发，linkedin于2010年贡献给了apache基金会并成为顶级开源项目。kafka是一种快速、可扩展的、设计内在就是分布式的，分区的和可复制的提交日志服务。
注意:kafka并没有遵循jms规范（），它只提供了发布和订阅通讯方式。
kafka核心相关名称broker：kafka节点，一个kafka节点就是一个broker，多个broker可以组成一个kafka集群
topic：一类消息，消息存放的目录即主题，例如page view日志、click日志等都可以以topic的形式存在，kafka集群能够同时负责多个topic的分发
massage： kafka中最基本的传递对象。
partition：topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。kafka里面实现分区，一个broker就是表示一个区域。
segment：partition物理上由多个segment组成，每个segment存着message信息
producer : 生产者，生产message发送到topic
consumer : 消费者，订阅topic并消费message, consumer作为一个线程来消费
consumer group：消费者组，一个consumer group包含多个consumer
offset：偏移量，理解为消息 partition 中消息的索引位置
主题和队列的区别：
队列是一个数据结构，遵循先进先出原则
kafka集群安装每台服务器上安装jdk1.8环境
安装zookeeper集群环境
安装kafka集群环境
运行环境测试
安装jdk环境和zookeeper这里不详述了。
kafka为什么依赖于zookeeper：kafka会将mq信息存放到zookeeper上，为了使整个集群能够方便扩展，采用zookeeper的事件通知相互感知。
kafka集群安装步骤：
1、下载kafka的压缩包
2、解压安装包
tar -zxvf kafka_2.11-1.0.0.tgz
3、修改kafka的配置文件 config/server.properties
配置文件修改内容：
zookeeper连接地址：zookeeper.connect=192.168.1.19:2181
监听的ip，修改为本机的iplisteners=plaintext://192.168.1.19:9092
kafka的brokerid，每台broker的id都不一样broker.id=0
4、依次启动kafka
./kafka-server-start.sh -daemon config/server.properties
kafka使用kafka文件存储topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是producer生成的数据。producer生成的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，kafka采取了分片和索引机制，将每个partition分为多个segment，每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号。
例如：执行命令新建一个主题，分三个区存放放在三个broker中：
./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic kaico
一个partition分为多个segment
.log 日志文件
.index 偏移量索引文件
.timeindex 时间戳索引文件
其他文件（partition.metadata，leader-epoch-checkpoint）
springboot整合kafkamaven依赖
<dependencies> <!-- springboot集成kafka --> <dependency> <groupid>org.springframework.kafka</groupid> <artifactid>spring-kafka</artifactid> </dependency> <!-- springboot整合web组件 --> <dependency> <groupid>org.springframework.boot</groupid> <artifactid>spring-boot-starter-web</artifactid> </dependency> </dependencies>
yml配置
# kafkaspring: kafka: # kafka服务器地址(可以多个)# bootstrap-servers: 192.168.212.164:9092,192.168.212.167:9092,192.168.212.168:9092 bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094 consumer: # 指定一个默认的组名 group-id: kafkagroup1 # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费 # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据 # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常 auto-offset-reset: earliest # key/value的反序列化 key-deserializer: org.apache.kafka.common.serialization.stringdeserializer value-deserializer: org.apache.kafka.common.serialization.stringdeserializer producer: # key/value的序列化 key-serializer: org.apache.kafka.common.serialization.stringserializer value-serializer: org.apache.kafka.common.serialization.stringserializer # 批量抓取 batch-size: 65536 # 缓存容量 buffer-memory: 524288 # 服务器地址 bootstrap-servers: www.kaicostudy.com:9092,www.kaicostudy.com:9093,www.kaicostudy.com:9094
生产者
@restcontrollerpublic class kafkacontroller { /** * 注入kafkatemplate */ @autowired private kafkatemplate<string, string> kafkatemplate; /** * 发送消息的方法 * * @param key * 推送数据的key * @param data * 推送数据的data */ private void send(string key, string data) { // topic 名称 key data 消息数据 kafkatemplate.send("kaico", key, data); } // test 主题 1 my_test 3 @requestmapping("/kafka") public string testkafka() { int imax = 6; for (int i = 1; i < imax; i++) { send("key" + i, "data" + i); } return "success"; }}
消费者
@componentpublic class topickaicoconsumer { /** * 消费者使用日志打印消息 */ @kafkalistener(topics = "kaico") //监听的主题 public void receive(consumerrecord<?, ?> consumer) { system.out.println("topic名称:" + consumer.topic() + ",key:" + consumer.key() + "," + "分区位置:" + consumer.partition() + ", 下标" + consumer.offset()); //输出key对应的value的值 system.out.println(consumer.value()); }}
以上就是java分布式之kafka消息队列实例分析的详细内容。