不背锅运维：消息队列概念、kafka入门、Kafka Golang客户端库

2023-03-15 23:15:30 浏览数 (2)

消息队列是什么

消息队列是一种在应用程序之间进行通信的技术，允许将消息从一个应用程序发送到另一个应用程序，而无需明确的连接这些应用程序。消息队列中的消息被存储在一种称为队列的数据结构中，这些消息在队列中保留，直到被消费者接收。这使得消息的发送者和接收者能够异步地通信，而不必等待对方的响应，从而提高了系统的可伸缩性和弹性。消息队列还可以通过实现各种模式（例如发布/订阅模式、请求/响应模式等）来支持不同类型的应用程序通信。

消息队列的关键概念

消息队列中的关键概念包括：

  1. 消息：要传递的数据或信息。
  2. 队列：用于存储消息的数据结构，具有先进先出（FIFO）的特性。
  3. 生产者：向消息队列发送消息的应用程序。
  4. 消费者：从消息队列接收消息的应用程序。
  5. 绑定（Binding）：将一个消息队列绑定到一个交换机上，以确保消息被路由到正确的队列。
  6. 交换机（Exchange）：接收来自生产者的消息并将其路由到一个或多个队列中。
  7. 路由键（Routing Key）：用于将消息从交换机路由到正确的队列。

“

这些概念组成了消息队列的核心，使得生产者和消费者能够异步地通信，从而提高了系统的可伸缩性和弹性。

”

消息队列的应用场景

消息队列的应用场景非常广泛，以下是其中一些常见的应用场景：

  1. 异步任务：将需要执行的任务放入消息队列中，由消费者异步地执行任务，提高系统的响应速度和并发性。
  2. 分布式系统：在分布式系统中，通过消息队列实现各个组件之间的异步通信，提高系统的可伸缩性和弹性。
  3. 应用解耦：将消息队列作为中间件，将各个应用程序解耦，避免直接依赖和影响，提高系统的可维护性和可扩展性。
  4. 日志收集：通过将日志消息发送到消息队列，使得日志的收集和分析可以异步地进行，避免直接影响业务处理。
  5. 消息通知：通过消息队列向用户发送通知消息，如短信、邮件等，提高系统的实时性和可靠性。
  6. 数据缓存：通过将热点数据缓存到消息队列中，减少系统的访问压力和响应时间。

“

总之，消息队列可以在各种分布式系统和异步场景中发挥作用，使得系统更加高效、灵活和可靠。

”

主流的消息队列有哪些

以下是一些主流的消息队列：

  1. Apache Kafka：Apache Kafka 是一种分布式发布-订阅消息系统，用于处理高吞吐量的实时数据。它是由 LinkedIn 开发的。
  2. RabbitMQ：RabbitMQ 是一个开源的消息代理，用于支持多种消息协议。它可以与多种编程语言一起使用，并且具有可扩展性和高可用性。
  3. Apache ActiveMQ：Apache ActiveMQ 是一个开源的消息代理，它实现了 JMS（Java 消息服务）规范，支持多种传输协议和消息格式。
  4. Amazon SQS：Amazon SQS 是亚马逊提供的一种完全托管的消息队列服务，它具有高可用性和可扩展性。
  5. Microsoft Azure Service Bus：Microsoft Azure Service Bus 是微软提供的一种完全托管的消息队列服务，支持多种协议和语言，具有高可用性和可扩展性。
  6. Google Cloud Pub/Sub：Google Cloud Pub/Sub 是 Google 提供的一种完全托管的消息队列服务，用于处理大量的实时数据流。

主流的消息队列对比

以下是一些常见的消息队列系统的对比：

  1. RabbitMQ：RabbitMQ 是一个流行的 AMQP（高级消息队列协议）消息代理。它具有良好的性能，可靠性和稳定性，并支持多种协议。RabbitMQ 是一个成熟的消息队列系统，具有广泛的社区支持和丰富的功能。它的管理界面易于使用，可用于监视和管理队列。
  2. Apache Kafka：Apache Kafka 是一个高性能的分布式流处理系统和消息队列平台。它设计用于处理大量的实时数据，并提供可扩展性，高吞吐量和低延迟的特性。Kafka 的消息模型是基于发布/订阅模式的，并支持多个消费者组，可以实现高效的消息分发和负载均衡。
  3. ActiveMQ：Apache ActiveMQ 是一个流行的开源消息代理，支持多种传输协议和编程语言，并提供高可用性和可扩展性的特性。它支持 JMS（Java 消息服务）规范，并提供可靠的消息传递和事务支持。
  4. Redis：Redis 是一种内存数据库，但它也可以用作消息队列。Redis 的消息队列模型是基于发布/订阅模式的，并支持多种数据结构和高级功能。Redis 的消息队列性能较高，但可靠性和持久性取决于其配置和使用方式。
  5. Apache Pulsar：Apache Pulsar 是一个开源的分布式流处理系统和消息队列平台，具有高可用性和可扩展性。它支持多种协议和编程语言，并提供高效的消息传递和低延迟的特性。Pulsar 的消息模型是基于发布/订阅和队列模式的，并支持多个消费者组和顺序消息。

“

总的来说，每个消息队列系统都有自己的优缺点和适用场景，具体选择取决于实际需求和条件。

”

主流的消息队列应用场景

不同的消息队列适用于不同的应用场景。以下是一些主流的消息队列的应用场景：

  1. RabbitMQ：
  • 高吞吐量的消息队列
  • 多种语言客户端库支持
  • 支持多种消息协议
  • 支持复杂的路由规则
  • 支持消息确认机制
  • 适合任务队列、日志处理、消息通信等场景
  1. Kafka：
  • 高吞吐量的消息队列
  • 分布式的设计，支持高可用和水平扩展
  • 支持消息的持久化和多副本备份
  • 支持批量发送和消费消息
  • 适合日志收集、流处理、消息通信等场景
  1. ActiveMQ：
  • 支持多种协议，如AMQP、STOMP、OpenWire等
  • 支持多种消息模式，如点对点、发布订阅等
  • 支持事务、消息确认等机制
  • 支持集群和主从模式
  • 适合消息通信、任务队列、日志处理等场景
  1. RocketMQ：
  • 高吞吐量的消息队列
  • 支持分布式和高可用架构
  • 支持多种消息协议
  • 支持批量发送和消费消息
  • 支持事务消息和消息轨迹
  • 适合大规模分布式系统、电商场景、金融场景等
  1. Redis：
  • 支持多种数据结构和多种操作，如list、pub/sub等
  • 支持持久化和复制
  • 支持Lua脚本扩展
  • 适合高速缓存、消息通信、任务队列等场景
  1. Pulsar：
  • 适合大规模分布式系统、物联网等场景
  • 支持消息的持久化和多副本备份
  • 支持多租户、多协议和多种语言客户端
  • 支持流处理和事务
  • 支持动态扩展和缩减

“

总的来说，不同的消息队列适用于不同的场景和需求，需要根据具体的业务需求选择合适的消息队列。

”

Kafka入门实战

关键概念

Kafka是一个分布式的流处理平台，常用于高吞吐量的数据管道和实时流数据处理。以下是Kafka的关键概念：

  1. Topic（主题）：Kafka中的消息都被发布到topic，一个topic可以被认为是一个数据源，也可以被认为是一个消息的分类。
  2. Partition（分区）：每个topic可以被分为多个partition，每个partition可以存储特定数量的消息。每个partition都有一个唯一的标识符（partition id）。
  3. Offset：每个partition中的每个消息都会被分配一个唯一的offset，它是该消息在partition中的唯一标识符。
  4. Producer（生产者）：负责将消息发布到指定的topic。
  5. Consumer（消费者）：消费者订阅了一个或多个topic，并处理被发布到这些topic的消息。
  6. Consumer Group：一组消费者可以组成一个消费者组，这些消费者一起消费一个topic中的所有partition，每个partition只能由一个消费者组中的消费者进行消费。
  7. Broker：Kafka集群中的每个节点都称为broker，负责接收和处理消息，一个Kafka集群可以由多个broker组成。
  8. ZooKeeper：Kafka使用ZooKeeper来维护集群的元数据，如broker的状态、topic和partition的状态等。

“

ZooKeeper是一个开源的分布式协调服务，用于维护配置信息、命名、提供分布式同步和提供组服务等功能。它被设计为高性能、高可用、高扩展性的分布式协调服务，可以使分布式应用程序更加简单和可靠。ZooKeeper采用ZAB协议（ZooKeeper Atomic Broadcast）实现主从复制，确保数据的强一致性。它的应用场景包括分布式锁、配置管理、服务发现、集群管理等。

”

Kafka集群搭建

  1. 添加名称和IP的映射
代码语言：txt复制
cat >> /etc/hosts << EOF
192.168.11.247 kafka01
192.168.11.248 kafka02
192.168.11.249 kafka03
EOF
  1. 关闭防火墙
代码语言：txt复制
systemctl stop firewalld
systemctl disable firewalld
  1. 创建普通用户
代码语言：txt复制
groupadd kafka
useradd -g kafka kafka
  1. 关闭 SELinux
代码语言：txt复制
setenforce 0 # 临时关闭
sed -i 's/SELINUX=enforcing/SELINUX=disabled/g' /etc/selinux/config # 永久关闭
  1. 安装java 1.8环境
代码语言：txt复制
# 可联网时使用Yum安装
yum install java-1.8.0-openjdk -y

[root@kafka01 ~]# java -version
openjdk version "1.8.0_362"
OpenJDK Runtime Environment (build 1.8.0_362-b08)
OpenJDK 64-Bit Server VM (build 25.362-b08, mixed mode)
[root@kafka01 ~]# 

“

注意：接下来的操作请切换到普通用户kafka下面进行操作

”

  1. 下载Kafka和ZooKeeper
代码语言：txt复制
wget https://archive.apache.org/dist/kafka/2.7.2/kafka_2.13-2.7.2.tgz
wget https://downloads.apache.org/zookeeper/zookeeper-3.7.1/apache-zookeeper-3.7.1-bin.tar.gz
  1. 解压Kafka和ZooKeeper
代码语言：txt复制
tar -zf kafka_2.13-2.7.2.tgz
tar -zxf apache-zookeeper-3.7.1-bin.tar.gz
  1. 配置ZooKeeper 在ZooKeeper中，我们需要创建一个配置文件，该文件应包含ZooKeeper集群中所有节点的IP地址。

在ZooKeeper的根目录中，创建一个名为conf的文件夹，并在其中创建一个名为zoo.cfg的文件。

在zoo.cfg中添加以下内容：

代码语言：txt复制
tickTime=2000
dataDir=/data/zookeeperData
clientPort=2181
initLimit=5
syncLimit=2
# server.1=IP_ADDRESS_NODE_1:2888:3888
# server.2=IP_ADDRESS_NODE_2:2888:3888
# server.3=IP_ADDRESS_NODE_3:2888:3888
server.1=192.168.11.247:2888:3888
server.2=192.168.11.248:2888:3888
server.3=192.168.11.249:2888:3888

将“IP_ADDRESS_NODE_1”、“IP_ADDRESS_NODE_2”和“IP_ADDRESS_NODE_3”替换为每个ZooKeeper节点的IP地址。

  1. 创建myid文件 分别在每台节点的/data/zookeeperData目录下创建myid，并写入编号，编号只能为数字，编号对应zoo.cfg配置中的server.1、server.2、server.3。下面是在节点1上的配置：
代码语言：txt复制
cd zookeeperData/
echo 1 > myid # 对应节点1

“

另外两个节点请根据实际情况调整配置。

”

  1. 启动ZooKeeper
代码语言：txt复制
 cd apache-zookeeper-3.7.1-bin
./bin/zkServer.sh start
  1. 配置Kafka 分别在每台节点的Kafka的根目录中，创建一个名为config的文件夹，并在其中创建一个名为server.properties的文件（如果有则直接编辑即可） 这是在节点1的server.properties中添加以下内容：
代码语言：txt复制
broker.id=1
listeners=listener1://192.168.11.247:9092
log.dirs=/data/kafka-logs
# zookeeper.connect=IP_ADDRESS_NODE_1:2181,IP_ADDRESS_NODE_2:2181,IP_ADDRESS_NODE_3:2181
zookeeper.connect=192.168.11.247:2181,192.168.11.248:2181,192.168.11.249:2181

将“IP_ADDRESS_NODE_1”、“IP_ADDRESS_NODE_2”和“IP_ADDRESS_NODE_3”替换为每个ZooKeeper节点的IP地址。

  • broker.id：每台节点都需要配置唯一的broker.id，以便Kafka能够正确地识别和管理节点。broker.id是一个整数，用于标识Kafka集群中的每个节点。
  • listeners：在Kafka集群中，listeners参数用于配置Kafka节点侦听客户端请求的地址和端口号。每台节点可能有多个listeners参数，以便可以从多个地址或端口号接收客户端请求。

“

另外两个节点请根据实际情况调整配置。

”

  1. 每台节点设置并行垃圾回收线程的数量 打开 Kafka 启动脚本kafka-server-start.sh，在文件中找到包含 KAFKA_HEAP_OPTS 的行，在 KAFKA_HEAP_OPTS 变量中添加 -XX:ParallelGCThreads 选项，例如 -XX:ParallelGCThreads=4，其中 4 表示您的系统有 4 个处理器。如果您不确定处理器数量，可以使用 Runtime.getRuntime().availableProcessors() 命令来查询。

例如，如果我的系统有 1 个处理器，您可以将 KAFKA_HEAP_OPTS 设置为：

代码语言：txt复制
export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G -XX:ParallelGCThreads=1"
  1. 在每个节点上启动Kafka
代码语言：txt复制
cd kafka_2.13-2.7.2
nohup bin/kafka-server-start.sh config/server.properties &

注意：在生产环境中，建议使用系统服务启动Kafka。

Kafka基本操作

  1. 创建一个Kafka主题
代码语言：txt复制
bin/kafka-topics.sh --create --zookeeper 192.168.11.247:2181 --replication-factor 3 --partitions 5 --topic my_topic

其中，my_topic是您要创建的主题的名称。在这个命令中，我们指定了主题的复制因子和分区数。replication-factor指定了主题的副本数，通常设置为大于1的值以实现数据冗余和高可用性。partitions指定了主题的分区数，这将决定Kafka如何在不同的消费者之间分配数据。

  1. 查看主题
代码语言：txt复制
bin/kafka-topics.sh --list --bootstrap-server 192.168.11.247:9092
  1. 删除主题
代码语言：txt复制
bin/kafka-topics.sh --zookeeper 192.168.11.247:2181 --delete --topic my_topic

如果 Kafka 配置中没有启用 delete.topic.enable 参数，那么主题的删除操作不会生效。等待一段时间，直到所有 Kafka 服务器都确认主题已被删除。可以使用以下命令检查主题是否已被删除：

代码语言：txt复制
bin/kafka-topics.sh --zookeeper 192.168.11.247:2181 --list

“

请注意，在生产环境中，删除主题时需要格外谨慎。删除主题将永久删除所有与该主题相关的消息和元数据。在删除主题之前，请确保备份了所有必要的数据并已通知所有相关方。

”

  1. 查看 Kafka 主题详情
代码语言：txt复制
bin/kafka-topics.sh --describe --bootstrap-server 192.168.11.247:9092 --topic my_topic
  1. 发送消息到 Kafka 主题
代码语言：txt复制
bin/kafka-console-producer.sh --broker-list 192.168.11.247:9092 --topic my_topic

输入以上命令后，会进入一个交互式的命令行界面。在该界面中，每行输入的文本将被作为一条消息发送到指定的主题中。按下 Ctrl C 即可退出该命令行工具。

在上述命令中，192.168.11.247:9092 是 Kafka 集群其中一个节点的地址，my-topic 是主题名称。如果 Kafka 集群有多个节点，则可以用逗号分隔的方式指定多个 Kafka 服务器地址，例如 --broker-list kafka1:9092,kafka2:9092,kafka3:9092。

除了 kafka-console-producer 工具，也可以在编程语言中使用 Kafka 客户端 API 发送消息到 Kafka 主题。

  1. 从 Kafka 主题中读取消息
代码语言：txt复制
bin/kafka-console-consumer.sh --bootstrap-server 192.168.11.247:9092 --topic my_topic

输入以上命令后，该命令行工具将从指定的主题中读取消息，并输出到命令行界面中。如果不指定 --from-beginning 参数，则该命令行工具将从最新的消息开始读取；如果指定了 --from-beginning 参数，则该命令行工具将从最早的消息开始读取。

  1. 查看当前 Kafka 服务器的健康状况
代码语言：txt复制
bin/kafka-broker-api-versions.sh --bootstrap-server 192.168.11.247:9092

消费者组

在 Kafka 中，消费者组是一组具有相同 Group ID 的消费者。消费者组可以订阅一个或多个主题，并共同消费这些主题的消息。每个消费者组中的消费者可以独立地消费消息，因此 Kafka 允许分布式处理消息。

当一个消息发送到一个订阅了该主题的消费者组时，Kafka 将该消息发送到组中的一个消费者。如果组中有多个消费者，则 Kafka 会采用一些算法来确定哪个消费者将接收消息，例如轮询、范围和散列等算法。一旦一个消费者接收到消息并开始处理它，其他消费者将无法接收该消息。这样可以确保消息仅被消费者组中的一个消费者处理，从而避免了重复处理消息的问题。

使用消费者组的好处包括：

  • 支持并行消费：使用消费者组，多个消费者可以并行消费同一个主题的消息，从而提高消息处理能力。
  • 提高可靠性：当一个消费者出现故障或离线时，其他消费者可以接替它来处理消息。
  • 控制消费进度：使用消费者组，可以控制消费者消费消息的位置和进度，例如从特定的偏移量开始消费消息，或者从最新的消息开始消费。

需要注意的是，消费者组在 Kafka 中是一个重要的概念，对于理解和使用 Kafka 来说非常重要。同时，Kafka 还提供了一些工具和 API，用于管理和监控消费者组的状态和偏移量信息，以确保 Kafka 消费者组的可靠性和高效性。

  1. 创建主题
代码语言：txt复制
bin/kafka-topics.sh --create --zookeeper 192.168.11.247:2181 --replication-factor 3 --partitions 5 --topic my_topic
  1. 启动一个消费者组并订阅该主题my_topic
代码语言：txt复制
bin/kafka-console-consumer.sh --bootstrap-server 192.168.11.247:9092 --topic my_topic --group test_group

这将启动一个名为“test_group”的消费者组，并使用“my_topic”主题进行订阅。每个新消息都将被发送到所有已连接的消费者。

  1. 在另一个终端窗口中使用生产者来向主题发送数据
代码语言：txt复制
bin/kafka-console-producer.sh --broker-list 192.168.11.247:9092 --topic my_topic

这将启动一个生产者，它将等待输入要发送到“my_topic”主题的消息。可以随时发送一些测试消息来测试消费者组是否按预期工作。

  1. 添加另一个消费者到同一组中，可以打开另一个终端窗口并使用相同的命令启动消费者
代码语言：txt复制
bin/kafka-console-consumer.sh --bootstrap-server 192.168.11.247:9092 --topic my_topic --group test_group

这将启动另一个消费者，并将其添加到名为“test_group”的消费者组中。现在，每个新消息都将被发送到这两个消费者，它们将共同处理工作负载。

“

请注意，您可以在生产者和消费者之间轻松切换，并尝试不同的组合以测试您的Kafka集群。此外，Kafka消费者组具有更高级的功能，如手动分配分区，重新平衡等，这些功能可以使用Kafka API进行实现。

”

kafka主题有多个分区的发送和读取机制

在 Kafka 主题中有多个分区的情况下，如果在发送消息时未指定分区，则 Kafka 会根据生产者的默认分区策略来确定将消息发送到哪个分区。如果在消费者端使用 kafka-console-consumer.sh 命令行工具来读取消息，并且未指定消费者要读取的分区，则 Kafka 将采用默认的分区分配策略，该策略会根据消费者组和主题的分区数来分配分区。Kafka 提供了几种分配策略，包括轮询、范围、散列等。默认情况下，使用轮询策略。

例如，如果您有一个主题，该主题有三个分区，并且有两个消费者加入同一消费者组并订阅该主题，则每个消费者将被分配到一个分区，并开始消费该分区中的消息。如果有第三个消费者加入消费者组，则该消费者将一直处于空闲状态，因为已经有两个消费者处理了所有的分区。

如果您在消费者端使用 kafka-console-consumer.sh 命令行工具来读取消息，并且想要指定要读取的分区，则可以使用 --partition 参数来指定要读取的分区。例如，使用以下命令来读取名为 test-topic 的主题的第 0 个分区中的消息：

代码语言：txt复制
bin/kafka-console-consumer.sh --bootstrap-server 192.168.11.247:9092 --topic test-topic --partition 0

“

总之，建议在 Kafka 主题中使用分区时，始终显式指定要发送和读取的分区，以确保消息在各个分区之间均匀分布，并避免消费者之间的负载不平衡。

”

Kafka Go客户端库

常用的Kafka Go客户端库

  1. Sarama：Sarama是一个使用Go编写的Kafka客户端库，提供了一系列API以简化与Kafka的交互。它支持各种Kafka功能，如生产者，消费者，管理员等。它还提供了一些高级功能，例如事务，压缩和TLS支持。
  2. Confluent-kafka-go：Confluent-kafka-go是一个由Confluent公司维护的Kafka客户端库。它提供了一系列API以连接Kafka集群并进行生产者和消费者操作。它支持TLS，SASL和Kerberos身份验证。
  3. Shopify/sarama：Shopify/sarama是一个简单易用的Kafka客户端库，支持Kafka 0.8.2及以上版本。它支持高吞吐量和低延迟，具有高度可配置性。它还提供了一些高级功能，例如Kafka消息的压缩和批处理。
  4. Segmentio/kafka-go：Segmentio/kafka-go是一个基于Go语言的Kafka客户端库，支持Kafka 0.8版本及以上。它提供了高性能的生产者和消费者API，并支持TLS和SASL身份验证。
  5. Shopbrain/kafkawire：Shopbrain/kafkawire是一个轻量级的Kafka客户端库，它使用HTTP/2协议连接Kafka集群。它支持生产者和消费者API，提供简单易用的API，适用于处理少量数据的场景。

“

这些库都提供了一系列API以与Kafka交互，并具有不同的特性和用例，您可以根据自己的需求选择适合自己的库。

”

开始写代码

  1. 安装Kafka Go客户端库
代码语言：txt复制
go get github.com/Shopify/sarama
  1. 使用以下代码创建一个名为“test_topic”的主题，该主题有3个副本和6个分区
代码语言：txt复制
package main

import (
 "fmt"

 "github.com/Shopify/sarama"
)

func main() {
 // 创建Kafka管理员客户端
 admin, err := sarama.NewClusterAdmin([]string{"192.168.11.247:9092"}, sarama.NewConfig())
 if err != nil {
  panic(err)
 }
 defer admin.Close()

 // 要创建的主题的名称、分区数和副本数
 topic := "test_topic"
 partitions := int32(6)
 replicationFactor := int16(3)

 // 创建主题配置
 topicConfig := &sarama.TopicDetail{
  NumPartitions:     partitions,
  ReplicationFactor: replicationFactor,
  ConfigEntries:     map[string]*string{},
 }

 // 使用管理员客户端的CreateTopic函数创建主题
 err = admin.CreateTopic(topic, topicConfig, false)
 if err != nil {
  panic(err)
 }

 fmt.Println("Topic created successfully")
}
  1. 创建一个生产者并向"test_topic"主题发送消息
代码语言：txt复制
package main

import (
 "fmt"

 "github.com/Shopify/sarama"
)

func main() {
 // 创建Kafka生产者配置
 config := sarama.NewConfig()
 config.Producer.RequiredAcks = sarama.WaitForAll
 config.Producer.Retry.Max = 5
 config.Producer.Return.Successes = true

 // 创建Kafka生产者
 producer, err := sarama.NewSyncProducer([]string{"192.168.11.247:9092"}, config)
 if err != nil {
  panic(err)
 }
 defer producer.Close()

 // 要发送的消息
 message := &sarama.ProducerMessage{
  Topic: "test_topic",
  Value: sarama.StringEncoder("hello, kafka"),
 }

 // 使用生产者的SendMessage函数发送消息
 partition, offset, err := producer.SendMessage(message)
 if err != nil {
  panic(err)
 }

 fmt.Printf("Message sent to partition %d at offset %dn", partition, offset)
}
  1. 创建一个消费者，从“test_topic”主题中消费消息
代码语言：txt复制
package main

import (
 "fmt"
 "os"
 "os/signal"
 "syscall"

 "github.com/Shopify/sarama"
)

func main() {
 // 创建Kafka消费者配置
 config := sarama.NewConfig()
 config.Consumer.Return.Errors = true

 // 创建Kafka消费者
 consumer, err := sarama.NewConsumer([]string{"192.168.11.247:9092"}, config)
 if err != nil {
  panic(err)
 }
 defer consumer.Close()

 // 消费的主题和分区
 topic := "test_topic"
 partition := int32(0)

 // 使用消费者的Consume函数消费消息
 partitionConsumer, err := consumer.ConsumePartition(topic, partition, sarama.OffsetNewest)
 if err != nil {
  panic(err)
 }
 defer partitionConsumer.Close()

 // 处理消费的消息
 signals := make(chan os.Signal, 1)
 signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)

 for {
  select {
  case msg := <-partitionConsumer.Messages():
   fmt.Printf("Received message: Topic=%s, Partition=%d, Offset=%d, Key=%s, Value=%sn",
    msg.Topic, msg.Partition, msg.Offset, string(msg.Key), string(msg.Value))
  case err := <-partitionConsumer.Errors():
   fmt.Println("Error while consuming partition:", err)
  case <-signals:
   fmt.Println("Interrupt signal received, shutting down consumer...")
   return
  }
 }
}

注意：在上面的例子中，生产者没有指定往哪个分区发消息，消费者也没有指定从哪个分区读取消息，那么机制是怎样？

Kafka的生产者在发送消息时可以不指定分区，这种情况下，Kafka会使用默认的分区策略来为消息选择一个分区。默认的分区策略是基于消息的key值进行哈希计算，从而确定消息应该被发送到哪个分区中。

如果消息没有key值，那么Kafka会使用轮询的方式将消息依次发送到每个可用的分区中，以实现负载均衡。

对于消费者来说，当不指定分区时，Kafka会将消费者分配给所有可用分区的某些分区，以使消费者能够消费所有分配给它的分区的消息。这个过程叫做分区分配。消费者可以通过指定消费者组来协调多个消费者之间的分区分配。如果消费者组中有多个消费者，则Kafka会将所有分区均匀地分配给每个消费者，以实现负载均衡。当消费者加入或离开消费者组时，Kafka会重新分配分区以确保负载均衡。

总的来说，Kafka的生产者和消费者通过默认的分区策略和分区分配机制来实现自动负载均衡，同时又能够保证数据的可靠性和有序性。

本文转载于WX公众号：不背锅运维（喜欢的盆友关注我们）：https://mp.weixin.qq.com/s/SPLs4wv6XHWRIoJVjb8qZg

zookeeper kafka ip partition 消息队列

0 人点赞