Kafka，RabbitMQ，RockedMQ實際應用大匯總1.Kafka
- 本文結合官網用例，記載了三大主流mq的實體以及實際運用，
- 本文不涉及相關環境的安裝與配置，涉及較為全面的代碼（包括組態檔及maven）
- 本文直接上代碼及用例，適合對mq已經學過一遍或了解過的同學進行學習&復習，可加入生產環境，
- 關于各種mq的介紹及對比可以參照我之前的文章
- 其實mq的設計思想都差不多，可以細細感受一下mq的設計理念以及基本的服務物件，
- 如果有問題，歡迎留言區或私信進行交流，
- 雖然已經挑重點代碼拿出來了，但是一篇文章里寫三個mq還是很長，因此還是分為三篇記錄吧，這是第一篇，主講kafka，
一：Kafka
-
基本/核心概念
-
Broker
Kafka的服務端程式，可以認為一個mq節點就是一個broker
broker存盤topic的資料 -
Producer生產者
創建訊息Message，然后發布到MQ中
該角色將訊息發布到Kafka的topic中 -
Consumer消費者:
消費佇列里面的訊息 -
ConsumerGroup消費者組
同個topic, 廣播發送給不同的group，一個group中只有一個consumer可以消費此訊息 -
Topic
每條發布到Kafka集群的訊息都有一個類別，這個類別被稱為Topic，主題的意思 -
Partition磁區
kafka資料存盤的基本單元，topic中的資料分割為一個或多個partition，每個topic至少有一個partition，是有序的
一個Topic的多個partitions, 被分布在kafka集群中的多個server上
消費者數量 <=小于或者等于Partition數量 -
Replication 副本（備胎）
同個Partition會有多個副本replication ，多個副本的資料是一樣的，當其他broker掛掉后，系統可以主動用副本提供服務
默認每個topic的副本都是1（默認是沒有副本，節省資源），也可以在創建topic的時候指定
如果當前kafka集群只有3個broker節點，則replication-factor最大就是3了，如果創建副本為4，則會報錯 -
ReplicationLeader、ReplicationFollower
Partition有多個副本，但只有一個replicationLeader負責該Partition和生產者消費者互動
ReplicationFollower只是做一個備份，從replicationLeader進行同步 -
ReplicationManager
負責Broker所有磁區副本資訊，Replication 副本狀態切換 -
offset
每個consumer實體需要為他消費的partition維護一個記錄自己消費到哪里的偏移offset
kafka把offset保存在消費端的消費者組里
-
-
kafka特點
-
多訂閱者
一個topic可以有一個或者多個訂閱者
每個訂閱者都要有一個partition，所以訂閱者數量要少于等于partition數量
高吞吐量、低延遲: 每秒可以處理幾十萬條訊息 -
高并發：幾千個客戶端同時讀寫
-
容錯性：多副本、多磁區，允許集群中節點失敗，如果副本資料量為n,則可以n-1個節點失敗
-
擴展性強：支持熱擴展
-
-
實體代碼(jdk11+kafka2.8)
- topic管理中心
package net.xdclass.kafkatest;
import org.apache.kafka.clients.admin.*;
import org.junit.jupiter.api.Test;
import java.util.*;
import java.util.concurrent.ExecutionException;
/**
* @Author NJUPT wly
* @Date 2021/8/14 12:15 上午
* @Version 1.0
*/
public class KafkaAdminTest {
private static final String TOPIC_NAME = "xdclass-sp-topic-1";
/**
* 設定admin客戶端
*/
public static AdminClient initAdminClient(){
Properties properties = new Properties();
properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
return AdminClient.create(properties);
}
@Test
public void createTopicTest(){
AdminClient adminClient = initAdminClient();
//指定磁區資料，副本數量
NewTopic newTopic = new NewTopic(TOPIC_NAME,5,(short) 1);
CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
try {
//future 等待創建
result.all().get();
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
@Test
public void listTopicTest() throws ExecutionException, InterruptedException {
AdminClient adminClient = initAdminClient();
ListTopicsOptions options = new ListTopicsOptions();
options.listInternal(true);
ListTopicsResult listTopicsResult = adminClient.listTopics(options);
Set<String> topics = listTopicsResult.names().get();
for (String name : topics){
System.out.println(name);
}
}
@Test
public void delTopicTest() throws ExecutionException, InterruptedException {
AdminClient adminClient =initAdminClient();
DeleteTopicsResult result = adminClient.deleteTopics(Collections.singletonList("xdclass-sp-topic"));
result.all().get();
}
@Test
public void detailTopicTest() throws ExecutionException, InterruptedException {
AdminClient adminClient = initAdminClient();
DescribeTopicsResult result = adminClient.describeTopics(Collections.singletonList(TOPIC_NAME));
Map<String,TopicDescription> stringTopicDescriptionMap = result.all().get();
Set<Map.Entry<String,TopicDescription>> entries = stringTopicDescriptionMap.entrySet();
entries.forEach((entry)-> System.out.println("name:"+entry.getKey()+",des"+entry.getValue()));
}
@Test
public void incrPartitionTest() throws ExecutionException, InterruptedException {
AdminClient adminClient = initAdminClient();
NewPartitions newPartitions = NewPartitions.increaseTo(5);
Map<String,NewPartitions> map = new HashMap<>();
map.put(TOPIC_NAME,newPartitions);
CreatePartitionsResult result = adminClient.createPartitions(map);
result.all().get();
}
}
- 消費者實體
package net.xdclass.kafkatest;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.Test;
import java.time.Duration;
import java.util.*;
/**
* @Author NJUPT wly
* @Date 2021/8/15 10:43 上午
* @Version 1.0
*/
public class KafkaConsumerTest {
private static final String TOPIC_NAME = "xdclass-sp-topic-1";
public static Properties getProperties(){
Properties properties = new Properties();
properties.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"xdclass-g-1");
properties.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");
// properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");
// properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,"1000");
properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"false");
properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
return properties;
}
@Test
public void simpleConsumerTest(){
Properties properties = getProperties();
KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
kafkaConsumer.subscribe(Collections.singleton(TOPIC_NAME));
while (true){
ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records){
System.err.printf("topic=%s,offset=%d,key=%s %n",record.topic(),record.offset(),record.key(),record.value());
}
// kafkaConsumer.commitSync();
if (!records.isEmpty()){
kafkaConsumer.commitAsync(new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
if (exception == null){
System.err.println("手動提交success"+offsets.toString());
} else {
System.err.println("手動提交fail"+offsets.toString());
}
}
});
}
}
}
}
- 生產者實體
package net.xdclass.kafkatest;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.clients.producer.internals.FutureRecordMetadata;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
/**
* @Author NJUPT wly
* @Date 2021/8/14 4:12 下午
* @Version 1.0
*/
public class KafkaProductTest {
private static final String TOPIC_NAME = "xdclass-sp-topic-1";
public static Properties getProperties(){
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"127.0.0.1:9092");
// properties.put("bootstrap.server","127.0.0.1:9092");
properties.put("acks","all");
properties.put("retries",0);
properties.put("batch",16384);
properties.put("linger.ms",1);
properties.put("buffer.memory",33554432);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
return properties;
}
@Test
public void testSend(){
Properties properties = getProperties();
Producer<String,String> producer = new KafkaProducer<String, String>(properties);
for (int i=0 ; i<3 ; i++){
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(TOPIC_NAME,"xdclass-key"+i,"xdclass-value"+i));
try {
RecordMetadata metadata = future.get();
System.out.println("發送狀態"+metadata.toString());
} catch (ExecutionException | InterruptedException e) {
e.printStackTrace();
}
}
producer.close();
}
@Test
public void testSendCallBack(){
Properties properties = getProperties();
Producer<String,String> producer = new KafkaProducer<String, String>(properties);
for (int i=0 ; i<3 ; i++){
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(TOPIC_NAME, "xdclass-key" + i, "xdclass-value" + i), new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e == null){
System.out.println("發送狀態"+recordMetadata.toString());
} else {
e.printStackTrace();
}
}
});
try {
RecordMetadata metadata = future.get();
System.out.println("發送狀態"+metadata.toString());
} catch (ExecutionException | InterruptedException e) {
e.printStackTrace();
}
}
producer.close();
}
@Test
public void testSendPartition(){
Properties properties = getProperties();
Producer<String,String> producer = new KafkaProducer<String, String>(properties);
for (int i=0 ; i<3 ; i++){
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(TOPIC_NAME,4,"xdclass-key"+i,"xdclass-value"+i));
try {
RecordMetadata metadata = future.get();
System.out.println("發送狀態"+metadata.toString());
} catch (ExecutionException | InterruptedException e) {
e.printStackTrace();
}
}
producer.close();
}
@Test
public void testSendP(){
Properties properties = getProperties();
properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG,"net.xdclass.kafkatest.config.PartitionerT");
Producer<String,String> producer = new KafkaProducer<String, String>(properties);
for (int i=0 ; i<3 ; i++){
Future<RecordMetadata> future = producer.send(new ProducerRecord<>(TOPIC_NAME, "xdclass" + i, "xdclass-value" + i), new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e == null){
System.out.println("發送狀態"+recordMetadata.toString());
} else {
e.printStackTrace();
}
}
});
try {
RecordMetadata metadata = future.get();
System.out.println("發送狀態"+metadata.toString());
} catch (ExecutionException | InterruptedException e) {
e.printStackTrace();
}
}
producer.close();
}
}
- 組態檔
logging:
config: classpath:logback.xml
spring:
kafka:
bootstrap-servers: 127.0.0.1:9092
producer:
retries: 1
batch-size: 16384
buffer-memory: 33554432
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
acks: all
transaction-id-prefix: xdclass-tran-
consumer:
auto-commit-interval: 1S
auto-offset-reset: earliest
enable-auto-commit: false
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
listener:
ack-mode: manual_immediate
concurrency: 4
- 以上為kafka原生api，上面除此之外，還有springboot整合spring-kafka，也是用于實際開發中的
- 監聽/消費者
package net.xdclass.kafkatest;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
/**
* @Author NJUPT wly
* @Date 2021/8/16 1:05 下午
* @Version 1.0
*/
@Component
public class LIstener {
@KafkaListener(topics = {"user.register.topic"},groupId = "xdclass-test-gp2")
public void Onmessage(ConsumerRecord<?,?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC)String topic){
System.out.println("消費"+record.topic()+"-"+record.partition()+"-"+record.value());
ack.acknowledge();
}
}
- controller（發送訊息）
package net.xdclass.kafkatest;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
/**
* @Author NJUPT wly
* @Date 2021/8/16 10:36 上午
* @Version 1.0
*/
@RestController
public class UserController {
private static final String TOPIC_NAME = "user.register.topic";
@Autowired
private KafkaTemplate<String,Object> kafkaTemplate;
@GetMapping("/api/v1/{num}")
public void sendMessage(@PathVariable("num")String num){
kafkaTemplate.send(TOPIC_NAME,"這是一個訊息"+num).addCallback(success->{
assert success != null;
String topic = success.getRecordMetadata().topic();
int partition = success.getRecordMetadata().partition();
long offset = success.getRecordMetadata().offset();
System.out.println(topic+partition+offset);
},failure->{
System.out.println("fail");
});
}
@GetMapping("/api/v1/tran")
public void sendMessage(int num){
kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback<String, Object, Object>() {
@Override
public Object doInOperations(KafkaOperations<String, Object> kafkaOperations) {
kafkaOperations.send(TOPIC_NAME,"這是一個訊息"+num);
return true;
}
});
}
}
- 完整pom
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.5.3</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>net.xdclass</groupId>
<artifactId>kafkatest</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>kafkatest</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>11</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>2.5.0</version>
</plugin>
</plugins>
</build>
</project>
轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/299487.html
標籤：其他