Message Queue Fundamentals

Definition

A message queue serves as an inter-process communication mechanism that enables asynchronous data exchange between applications. Messages represent the data units transmitted between systems, while the queue provides a reliable delivery mechanism ensuring data integrity throughout the transfer process.

Core Characteristics

Storage Mechanism

Messages reside in a buffered environment until the intended consumer retrieves them or explicitly removes them from the queue. This buffer-based approach decouples producers from consumers, allowing independent operation without direct coordination.

Asynchronous Processing

The queue introduces a layer of indirection between message producers and consumers. Producers dispatch messages without waiting for consumer acknowledgment, accumulating messages in the queue while consumers process them at their own pace.

Key Benefits

• Loose coupling between application components
• Data redundancy through message persistence
• Horizontal scalability for handling increased load
• Flexible message routing and processing
• Capability to handle traffic bursts
• System resilience with message recovery support
• Preserving message ordering within partitions
• Non-blocking communication patterns

Apache Kafka Architecture

Overview

Apache Kafka functions as a high-throughput distributed messaging system supporting publish-subscribe patterns. As an Apache Software Foundation project, it excels at processing streaming data in real-time across diverse use cases.

Core Terminology

Component	Description
Broker	Each server instance in a Kafka cluster operates as a broker, handling message storage and retrieval
Topic	A categorical channel where messages get published, serving as the logical grouping mechanism
Producer	Application responsible for publishing messages to brokers
Consumer	Application that subscribes to topics and processes incoming messages
Partition	Physical subdivision of a topic, enabling parallel processing and scalability
Consumer Group	Logical grouping of consumers sharing message processing workload
Message	Basic data unit transmitted through the system

Kafka Topology

Kafka employs a distributed architecture where topics get partitioned across multiple brokers, ensuring fault tolerance and parallel processing capabilities.

Apache ZooKeeper

Overview

ZooKeeper provides distributed coordination services essential for managing consensus, leader election, and configuration maintenance in clustered environments. It prevents race conditions and resource contention across distributed systems.

Operational Model

Leader Election Process

When nodes initialize, each registers itself with ZooKeeper. The election mechanism selects the node with the lowest identifier as the leader. The remaining nodes assume follower roles, maintaining standby capability for failover scenarios.

Failure Detection

When the current leader becomes unresponsive, ZooKeeper detects the session termination and triggers a new election cycle. The surviving nodes reassess and elect a replacement leader.

Recovery Procedure

Upon恢复正常，the former leader rejoins the cluster by registering with ZooKeeper. However, its node identifier now carries a higher value, preventing it from reclaiming leadership status.

ZooKeeper Roles

Role	Responsibilities
Leader	Initiates consensus votes, updates system state, coordinates operations
Follower	Handles client requests, returns responses, participates in election投票
Observer	Processes client requests by forwarding to leader, synchronizes state without voting rights, enhances scalability
Client	Initiates requests to ZooKeeper services

ZooKeeper Integration with Kafka

ZooKeeper manages several critical Kafka operations:

• Broker registration and health monitoring
• Topic metadata and partition assignment
• Producer load balancing across brokers
• Consumer group coordination
• Tracking partition-to-consumer mappings
• Storing consumer offset positions
• Consumer instance registration

Single-Node Kafka Setup

Environment: kafka1 (192.168.10.101)

ZooKeeper Installation

[root@kafka1 ~]# yum -y install java
[root@kafka1 ~]# tar zxvf apache-zookeeper-3.6.0-bin.tar.gz
[root@kafka1 ~]# mv apache-zookeeper-3.6.0-bin /etc/zookeeper
[root@kafka1 ~]# cd /etc/zookeeper/conf
[root@kafka1 ~]# mv zoo_sample.cfg zoo.cfg
[root@kafka1 ~]# vim zoo.cfg
dataDir=/etc/zookeeper/zookeeper-data

[root@kafka1 ~]# mkdir -p /etc/zookeeper/zookeeper-data
[root@kafka1 ~]# ./bin/zkServer.sh start
[root@kafka1 ~]# ./bin/zkServer.sh status

Kafka Installation

[root@kafka1 ~]# tar zxvf kafka_2.13-2.4.1.tgz
[root@kafka1 ~]# mv kafka_2.13-2.4.1 /etc/kafka
[root@kafka1 ~]# cd /etc/kafka/

[root@kafka1 kafka]# vim config/server.properties
log.dirs=/etc/kafka/kafka-logs   # line 60

[root@kafka1 kafka]# mkdir /etc/kafka/kafka-logs
[root@kafka1 kafka]# bin/kafka-server-start.sh config/server.properties &

# Verify ports are listening
[root@kafka1 kafka]# netstat -anpt | grep 2181
[root@kafka1 kafka]# netstat -anpt | grep 9092

Important: Always start ZooKeeper before Kafka, and shut down Kafka before stopping ZooKeeper.

Validation Commands

# Create a topic
./bin/kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test

# List available topics
./bin/kafka-topics.sh --list --zookeeper kafka1:2181

# Inspect topic details
./bin/kafka-topics.sh --describe --zookeeper kafka1:2181 --topic test

# Produce messages
./bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test

# Consume messages (open another terminal)
./bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test

# Remove a topic
./bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test

Multi-Node Kafka Cluster Deployment

Infrastructure

• kafka1: 192.168.10.101
• kafka2: 192.168.10.102
• kafka3: 192.168.10.103

Host Resolution

Configure /etc/hosts on all machines:

[root@kafka1 ~]# vim /etc/hosts
192.168.10.101 kafka1
192.168.10.102 kafka2
192.168.10.103 kafka3

ZooKeeper Cluster Configuration

Installation (identical across all nodes)

[root@kafka1 ~]# yum -y install java
[root@kafka1 ~]# tar zxvf apache-zookeeper-3.6.0-bin.tar.gz
[root@kafka1 ~]# mv apache-zookeeper-3.6.0-bin /etc/zookeeper
[root@kafka1 ~]# mkdir -p /etc/zookeeper/zookeeper-data

Configuration File

[root@kafka1 ~]# cd /etc/zookeeper/conf
[root@kafka1 ~]# mv zoo_sample.cfg zoo.cfg
[root@kafka1 ~]# vim zoo.cfg
dataDir=/etc/zookeeper/zookeeper-data
clientPort=2181
server.1=192.168.10.101:2888:3888
server.2=192.168.10.102:2888:3888
server.3=192.168.10.103:2888:3888

Node Identifier Setup

# On kafka1:
[root@kafka1 zookeeper-data]# echo 1 > myid

# On kafka2:
[root@kafka2 zookeeper-data]# echo 2 > myid

# On kafka3:
[root@kafka3 zookeeper-data]# echo 3 > myid

Startup Procedure

[root@kafka1 ~]# cd /etc/zookeeper
[root@kafka1 zookeeper]# ./bin/zkServer.sh start
[root@kafka1 zookeeper]# ./bin/zkServer.sh status

Kafka Cluster Configuration

Installation (identical across all nodes)

[root@kafka1 ~]# tar zxvf kafka_2.13-2.4.1.tgz
[root@kafka1 ~]# mv kafka_2.13-2.4.1 /etc/kafka

Broker Configuration

[root@kafka1 ~]# cd /etc/kafka/
[root@kafka1 kafka]# vim config/server.properties

# Broker identification (unique per node: 1, 2, 3)
broker.id=1

# Listener binding (node-specific IP)
listeners=PLAINTEXT://192.168.10.101:9092

# Message storage location
log.dirs=/etc/kafka/kafka-logs

# Partition count (must not exceed total nodes)
num.partitions=1

# ZooKeeper ensemble connection
zookeeper.connect=192.168.10.101:2181,192.168.10.102:2181,192.168.10.103:2181

Directory Preparation

[root@kafka1 kafka]# mkdir -p /etc/kafka/kafka-logs

Cluster Startup

Execute on all Kafka nodes:

[root@kafka1 kafka]# ./bin/kafka-server-start.sh config/server.properties &

If startup fails, clear the contents of /etc/kafka/kafka-logs and retry.

Cluster Testing

# Create topic (any node)
bin/kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 1 --partitions 1 --topic test

# List topics (query any ZooKeeper node)
bin/kafka-topics.sh --list --zookeeper kafka1:2181
bin/kafka-topics.sh --list --zookeeper kafka2:2181
bin/kafka-topics.sh --list --zookeeper kafka3:2181

# Publish test messages
bin/kafka-console-producer.sh --broker-list kafka1:9092 -topic test

# Consume messages
bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test

# Remove topic
bin/kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test