Apache Kafka 是分散式發布-訂閱訊息系統，在 kafka官網上對 kafka 的定義：一個分散式發布-訂閱訊息傳遞系統。它最初由LinkedIn公司開發，Linkedin於2010年貢獻給了Apache基金會並成為頂級開源專案。 Kafka是一種快速、可擴展的、設計內在就是分散式的，分區的和可複製的提交日誌服務。
注意:Kafka並沒有遵循JMS規範（），它只提供了發布和訂閱通訊方式。
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 中訊息的索引位置
主題和佇列的區別：
佇列是一個資料結構，遵循先進先出原則
每台伺服器上安裝jdk1.8環境
安裝Zookeeper叢集環境
安裝kafka叢集環境
執行環境測試
安裝jdk環境和zookeeper這裡不詳述了。
kafka為什麼依賴zookeeper：kafka會將mq資訊存放在zookeeper上，為了讓整個叢集能夠方便擴展，採用zookeeper的事件通知相互感知。
kafka叢集安裝步驟：
1、下載kafka的壓縮套件
2、解壓縮安裝套件
##tar -zxvf kafka_2.11 -1.0.0.tgz3、修改kafka的設定檔config/server.properties設定檔修改內容：
zookeeper.connect=192.168.1.19:2181
listeners=PLAINTEXT:// 192.168.1.19:9092
broker.id=0
./kafka-server-start.sh -daemon config/server.properties
#./kafka-topics.sh --create --zookeeper localhost: 2181 --replication-factor 1 --partitions 3 --topic kaico
<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># kafka
spring:
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@RestController
public 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";
}
}@Component
public 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訊息佇列實例分析的詳細內容。更多資訊請關注PHP中文網其他相關文章！
