Kafka集群安裝、配置和啟動
Kafka需要依賴zookeeper，并且自身集成了zookeeper，zookeeper至少需要3個節(jié)點(diǎn)保證集群高可用，下面是在單機(jī)linux下創(chuàng)建kafka3個節(jié)點(diǎn)偽集群模式。
1、下載包
下載地址：http://kafka.Apache.org/downloads
2、解壓包
tar -zxvf kafka_2.11-1.0.0.tgz
mv kafka_2.11-1.0.0 kafka1
mv kafka_2.11-1.0.0 kafka2
mv kafka_2.11-1.0.0 kafka3
3、創(chuàng)建ZK集群
修改ZK配置文件：kafka1-3/config/zookeeper.properties分別修改對應(yīng)的參數(shù)。
dataDir=/usr/local/kafka/zookeeper1
dataLogDir=/usr/local/kafka/zookeeper/log
clientPort=2181
maxClientCnxns=0
tickTime=2000
initLimit=100
syncLimit=5
server.1=127.0.0.1:2888:3888
server.2=127.0.0.1:4888:5888
server.3=127.0.0.1:6888:7888
/usr/local/kafka/zookeeper1-3目錄下分別創(chuàng)建myid文件，內(nèi)容對應(yīng)1~3
啟動ZK，分別進(jìn)行Kafka1-3目錄:
bin/zookeeper-server-start.sh config/zookeeper.properties &
啟動報文件失敗，需要手動創(chuàng)建文件目錄并賦予對應(yīng)的權(quán)限。
4、創(chuàng)建Kafka集群
配置文件：kafka1-3/config/server.properties分別修改對應(yīng)的參數(shù)。
broker.id=1
zookeeper.connect=localhost:2181,localhost:2182,localhost:2183
listeners=PLAINTEXT://192.168.12.11:9091
log.dirs=/tmp/kafka-logs-1
啟動Kafka，分別進(jìn)行Kafka1-3目錄:
bin/kafka-server-start.sh config/server.properties &
啟動報文件失敗，需要手動創(chuàng)建文件目錄并賦予對應(yīng)的權(quán)限。
5、集群測試
在kafka1上面發(fā)送消息：
bin/kafka-console-producer.sh --broker-list localhost:9091 --topic test
在kafka2、kafka3消費(fèi)消息：
bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic
Spring Boot 集成 Kafka 實(shí)戰(zhàn)
1、添加spring-kafka依賴
<spring-kafka.version>2.1.0.RELEASE</spring-kafka.version>
<!-- spring-kafka-->
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>${spring-kafka.version}</version>
</dependency>
2、添加Spring Boot的自動配置
自動配置類：
org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration
配置屬性類：
org.springframework.boot.autoconfigure.kafka.KafkaProperties
Spring:
kafka:
bootstrap-servers:
- 192.168.101.137:9091
- 192.168.101.137:9092
- 192.168.101.137:9093
producer:
retries: 0
batch-size: 16384
buffer-memory: 33554432
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
consumer:
group-id: foo
auto-offset-reset: earliest
enable-auto-commit: true
auto-commit-interval: 100
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
3、發(fā)送消息
@Autowired
private KafkaTemplate kafkaTemplate;
@GetMApping("/send")
public Object send(String msg) {
kafkaTemplate.send("test", "name", msg);
return "send ok";
}
4、接收消息
在任何bean里面，添加@KafkaListener，支持消息接收。
@KafkaListener(topics = "test")
public void processMessage(String content) {
logger.info("收到消息, topic:test, msg:{}", content);
}
5、參考資料
Spring Boot & Kafka官方文檔：
https://docs.spring.io/spring-boot/docs/current-SNAPSHOT/reference/htmlsingle/#boot-features-kafka
Spring for Apache Kafka官方文檔：
https://docs.spring.io/spring-kafka/reference/htmlsingle/
點(diǎn)擊了解更多鏈接關(guān)注我，分享更主流的JAVA技術(shù)～
