免费观看又色又爽又黄的小说免费_美女福利视频国产片_亚洲欧美精品_美国一级大黄大色毛片

如何使用消息隊列

本篇內容介紹了“如何使用消息隊列”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

做網站、成都網站設計的關注點不是能為您做些什么網站，而是怎么做網站，有沒有做好網站，給成都創新互聯一個展示的機會來證明自己，這并不會花費您太多時間，或許會給您帶來新的靈感和驚喜。面向用戶友好，注重用戶體驗，一切以用戶為中心。

1 概述

1.1 基本概念
1.1.1 Broker 代理

已發布的消息保存在一組服務器中，稱為Kafka集群。集群中的每個服務器都是一個Broker。

1.1.2 Topic 主題

通過Topic機制對消息進行分類，可以認為每個Topic就是一個隊列。

1.1.3 Partition 分區

每個Topic可以有多個分區，主要為了提高并發而設計。相同Topic下不同Partition可以并發接收消息，同時也能供消費者并發拉取消息。有多少Partition就有多少并發量。

在Kafka服務器上，分區是以文件目錄的形式存在的。每個分區目錄中，Kafka會按配置大小及配置周期將分區拆分成多個段文件(LogSegment)，每個段由三部分組成：

- 日志文件：*.log
- 位移索引文件：*.index
- 時間索引文件：*.timeindex

其中*.log用于存儲消息本身的數據內容，*.index存儲消息在文件中的位置(包括消息的邏輯offset和物理存儲offset)，*.timeindex存儲消息創建時間和對應邏輯地址的映射關系。

將分區拆分成多個段是為了控制存儲文件大小。可以很方便的通過操作系統mmap機制映射到內存中，提高寫入和讀取效率。同時還有一個好處就是，當系統要清除過期數據時，可以直接將過期的段文件刪除。

如果每個消息都要在index中保存位置信息，index文件自身大小也很容易變的很大。所以Kafka將index設計為稀疏索引來減小index文件的大小。

1.1.4 Replication 副本

消息冗余數量。不能超過集群中Broker的數量。

1.2 基本操作
1.2.1 Topic相關
# 創建Topic 
# --topic 主題名稱 避免使用[_]及[.]號
# --replication-factor 副本數量(不能超過broker節點數)
# --partitions 分區數量(并發)
./bin/kafka-topics.sh --create \
--topic UserDataQueue \
--replication-factor 3 \
--partitions 5 \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 查看Topic
./bin/kafka-topics.sh --list \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 修改Topic
# 刪除Topic
1.2.2 Message相關
# 發送消息
# --topic 指定目標Topic
./bin/kafka-console-producer.sh \
--topic UserDataQueue \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094

# 拉取消息
# --from-beginning 從頭開始(獲取現有的全量數據)
./bin/kafka-console-consumer.sh \
--topic UserDataQueue \
--bootstrap-server localhost:9092,localhost:9093,localhost:9094 \
--from-beginning

2 集群配置

Kafka集群依賴于Zookeeper。

2.1 Zookeeper配置及啟動
# 需要修改的參數

# the directory where the snapshot is stored.
dataDir=/kafka/zkdata
# the port at which the clients will connect
clientPort=2182
# 啟動
./bin/zookeeper-server-start.sh -daemon /kafka/zookeeper.properties
2.2 Kafka配置及啟動
# 需修改參數

# The id of the broker. This must be set to a unique integer for each broker.
broker.id=1  # 同一集群內ID必須唯一

# The address the socket server listens on. It will get the value returned from 
# java.net.InetAddress.getCanonicalHostName() if not configured.
#   FORMAT:
#     listeners = listener_name://host_name:port
#   EXAMPLE:
#     listeners = PLAINTEXT://your.host.name:9092
listeners=PLAINTEXT://localhost:9092  # 同一主機的話，端口號不能相同

# A comma separated list of directories under which to store log files
log.dirs=/kafka/data01  # 日志存儲目錄，需做隔離

# Zookeeper connection string (see zookeeper docs for details).
# This is a comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
# You can also append an optional chroot string to the urls to specify the
# root directory for all kafka znodes.
zookeeper.connect=localhost:2182  # Zookeeper連接地址，參見2.1 zk配置
# Kafka啟動

# broker-1
./bin/kafka-server-start.sh -daemon /kafka/server01.properties

# broker-2
./bin/kafka-server-start.sh -daemon /kafka/server02.properties

# broker-3
./bin/kafka-server-start.sh -daemon /kafka/server03.properties
2.3 Zookeeper可視化

PrettyZoo 是一個基于 Apache Curator 和 JavaFX 實現的 Zookeeper 圖形化管理客戶端。

由下圖可以看到，集群3個Broker均正常啟動。

如何使用消息隊列

2.4 Kafka可視化及監控
2.4.1 AKHQ

管理Topic，Topic消息，消費組等的Kafka可視化系統，相關文檔：https://akhq.io/

如何使用消息隊列

2.4.2 Kafka Eagle

一個簡單且高效的監控系統。相關文檔：http://www.kafka-eagle.org/index.html

如何使用消息隊列

Kafka Eagle 自帶監控大屏。

如何使用消息隊列

3 與Spring Boot集成

Spring Boot版本：2.4.4。

官方示例：https://github.com/spring-projects/spring-kafka/tree/main/samples

3.1 Spring Boot
3.1.1 添加依賴
implementation 'org.springframework.kafka:spring-kafka'
3.1.2 配置文件
spring:
  kafka:
    bootstrap-servers: localhost:9092,localhost:9093,localhost:9094
    producer:
      client-id: kfk-demo
      retries: 3
3.1.3 消息發送
@RestController
public class IndexController {

    @Autowired
    KafkaTemplate<Object, Object> kafkaTemplate;

    @GetMapping
    public String index() {
        int rdm = new Random().nextInt(1000);
        kafkaTemplate.send("UserDataQueue", new UserData("", rdm));
        return "hello world";
    }

    @GetMapping("index2")
    public String index2() {
		// 發送字符串方式
        kafkaTemplate.send("UserDataTopic", new Gson().toJson(new UserData("apple", 23)));
        return "ok";
    }
}
3.1.4 消息接收
@Component
@KafkaListener(
        id = "kfk-demo-userdata",
        topics = {"UserDataQueue"},
        groupId = "kfk-demo-group",
        clientIdPrefix = "kfk-demo-client")
public class KfkListener {

    @KafkaHandler
    public void process(@Payload UserData ud,
                        @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                        @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
        System.out.println(String.format("topic: %s, partition: %d, userData: %s", topic, partition, ud));
    }

    @KafkaHandler(isDefault = true)
    public void process(Object obj) {
        System.out.println(obj);
    }
}

// 接收字符串方式
@Slf4j
@Component
@KafkaListener(id = "kfk-demo2", topics = {"UserDataTopic"})
public class KfkUserDataTopicListener {

    @KafkaHandler
    public void process(String userDataStr) {
        UserData userData = new Gson().fromJson(userDataStr, UserData.class);
        log.info("username: {}, age: {}", userData.getUsername(), userData.getAge());
    }
}
3.1.5 Topic自動創建
@Configuration
public class KafkaConfig {

    @Bean
    public NewTopic userDataTopic() {
        return new NewTopic("UserDataTopic", 3, (short) 1);
    }
}

“如何使用消息隊列”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注創新互聯網站，小編將為大家輸出更多高質量的實用文章！

本文標題：如何使用消息隊列
當前路徑：http://m.newbst.com/article34/gseise.html

成都網站建設公司_創新互聯，為您提供企業建站、云服務器、建站公司、網站設計、標簽優化、定制開發

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯

小程序開發