Kafka Producer之拦截器
发表更新SpringCloud / Kafka3 分钟读完 (大约473个字)

Kafka Producer之拦截器

1. Producer流程

新建ProducerRecord类后，传入topic、key、value等数据构建Record之后，距离发送至kafka集群还需要经历若干过程。

  1. 拦截器列表，对数据进行过滤，更改等行为，处理异常不会导致流程终止。

  2. 获取Kafka集群元数据

  3. 对数据进行序列化

  4. 根据元数据选择分区和Broker

  5. 数据校验

  6. 进入数据发送缓存区，批次发送

  7. send

2. 代码测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
public class KafkaProducerInterceptorTest {
    public static void main(String[] args) throws InterruptedException {
        //创建producer
        HashMap<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //指定拦截器
        config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ValueInterceptorTest.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config);

        for (int i = 0; i < 10; i++) {
            //创建record
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                    "test1",
                    "key"+i,
                    "我是你爹"+i
            );
            //发送record
            producer.send(record);
            Thread.sleep(500);
        }

        //关闭producer
        producer.close();
    }
}

拦截器自定义类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
package org.dragon.producer;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

/**
 * 自定义value拦截器试验<br/>
 * 1. 实现接口
 * 2. 定义泛型
 * 3. 重写方法
 *
 *
 * @author mumu
 * @date 2024/07/15
 */
public class ValueInterceptorTest implements ProducerInterceptor<String, String> {

    /**
     * 发送数据时会调用这个方法<br/>
     * 让value复制2次
     *
     * @param producerRecord 生产者记录
     * @return {@link ProducerRecord}<{@link String}, {@link String}>
     */
    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        return new ProducerRecord<String, String>(producerRecord.topic(), producerRecord.key(), producerRecord.value() + producerRecord.value());
    }

    /**
     * 发送数据完毕，服务器返回的响应，会调用此方法。
     *
     * @param recordMetadata 记录元数据
     * @param e              e
     */
    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    /**
     * 生产者关闭，会调用此方法
     */
    @Override
    public void close() {

    }

    /**
     * 创建生产者对象时调用
     *
     * @param map 地图
     */
    @Override
    public void configure(Map<String, ?> map) {

    }
}

3. 测试结果

可以看到value是复制了2次，成功。

关于Kafka Topic分区和Replication分配的策略
发表更新SpringCloud / Kafka6 分钟读完 (大约937个字)

关于Kafka Topic分区和Replication分配的策略

1. Topic多分区

如图，是一个多分区Topic在Kafka集群中可能得分配情况。

P0-RL代表分区0，Leader副本。

这个Topic是3分区2副本的配置。分区尽量均匀分在不同的Broker上，分区的Follower副本尽量不和Leader在一个Broker上。

2. 理想的策略

假设有3个Topic在含有3个Broker的Kafka集群上。

Topic1有1个分区，2个副本。

Topic2有2个分区，2个副本。

Topic3有3个分区，2个副本。

可能如下图所示。不同颜色表示不同Topic。

似乎不是特别理想，我们再调整一下，如下图

不仅每个Broker的副本数一样了，更关键的是，并且每个Broker的主Leader副本也一样的。这样更适合负载均衡。

3. 实际的策略

我们使用Kafka tool，来以此创建上述3个Topic。

首先看test1

然后看test2

然后是test3

按照上面的信息，画出来的分配结果如下图

似乎并不和我们想的一样。

查看源码，Breadcrumbskafka/server-common/src/main/java/org/apache/kafka/admin/AdminUtils.java中一段代码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
private static Map<Integer, List<Integer>> assignReplicasToBrokersRackUnaware(int nPartitions,
                                                                                  int replicationFactor,
                                                                                  List<Integer> brokerList,
                                                                                  int fixedStartIndex,
                                                                                  int startPartitionId) {
        Map<Integer, List<Integer>> ret = new HashMap<>();
        int startIndex = fixedStartIndex >= 0 ? fixedStartIndex : RAND.nextInt(brokerList.size());
        int currentPartitionId = Math.max(0, startPartitionId);
        int nextReplicaShift = fixedStartIndex >= 0 ? fixedStartIndex : RAND.nextInt(brokerList.size());
        for (int i = 0; i < nPartitions; i++) {
            if (currentPartitionId > 0 && (currentPartitionId % brokerList.size() == 0))
                nextReplicaShift += 1;
            int firstReplicaIndex = (currentPartitionId + startIndex) % brokerList.size();
            List<Integer> replicaBuffer = new ArrayList<>();
            replicaBuffer.add(brokerList.get(firstReplicaIndex));
            for (int j = 0; j < replicationFactor - 1; j++)
                replicaBuffer.add(brokerList.get(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerList.size())));
            ret.put(currentPartitionId, replicaBuffer);
            currentPartitionId += 1;
        }
        return ret;
    }

例子（来自尚硅谷）

4. 如何自定义策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public class AdminTopicTest {
    public static void main(String[] args) {
        //定义kafka集群配置
        Map<String, Object> config = new HashMap<>();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");

        //创建Admin管理员对象
        Admin admin = Admin.create(config);

        //定义Topic属性
        HashMap<Integer, List<Integer>> map = new HashMap<>();
        // 分区0，Leader副本在3上，第二个副本在1上。
        map.put(0, Arrays.asList(3, 1));
        map.put(1, Arrays.asList(2, 3));
        map.put(2, Arrays.asList(1, 2));
        NewTopic test4 = new NewTopic("test2", map);


        //创建Topic
        CreateTopicsResult result = admin.createTopics(
                Arrays.asList(
                        test4
                )
        );

        admin.close();
    }
}

不过在手动分配时，确实需要了解每个broker的负载情况，以便做出更优的分配策略。你可以使用Kafka的AdminClient类来获取集群的状态信息

新增了一些新的封面
Docker Compose部署Kafka集群并在宿主机Windows连接开发
发表更新SpringCloud / Kafka9 分钟读完 (大约1383个字)

Docker Compose部署Kafka集群并在宿主机Windows连接开发

Docker for Windows 4.23.0
windows 11
Java 17

1. 常用参数

kafka容器常用参数如下

  • -e KAFKA_BROKER_ID=1：设置 Kafka broker 的 ID 为 1。每个 Kafka broker 都需要一个唯一的 ID。

  • -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181：指定 Kafka 连接到 Zookeeper 的地址，这里假设 Zookeeper 容器的名称为 zookeeper，并且它在 2181 端口监听。

  • -e ALLOW_PLAINTEXT_LISTENER=yes：允许 Kafka 使用纯文本监听器。即允许非加密的通信。

  • -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092：Kafka broker 实际监听在容器内的 0.0.0.0:9092 上。这意味着 Kafka 接受来自任何网络接口的连接。

  • -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092：指定 Kafka 广播其监听器地址，客户端将使用该地址连接到 broker。在这个例子中，Kafka 广播它在 localhost:9092 上监听。

  • KAFKA_LISTENER_SECURITY_PROTOCOL_MAP：指定 Kafka 使用的监听器协议映射。例如：PLAINTEXT:PLAINTEXT,SSL:SSL。

  • KAFKA_INTER_BROKER_LISTENER_NAME：指定 broker 间通信使用的监听器名称。例如：PLAINTEXT。

2. 理解参数和原理

KAFKA_LISTENERS是broker实际监听的地址。

KAFKA_ADVERTISED_LISTENERS是broker注册在zookeeper或者controller broker里面的元数据，当消费者或者生产者使用Bootstrap-Server去连接kafka集群时，集群会返回元数据等信息到客户端，客户端会根据每个broker提供的KAFKA_ADVERTISED_LISTENERS去连接对应的broker。

所以首先，集群之间，broker之间需要通信，所以每个kafka容器需要设置一个KAFKA_ADVERTISED_LISTENERS用于告诉别的容器如何连接到自己，如果容器都是处于同一bridge网络，那么直接使用容器名即可。

其次，我们想要在宿主机比如windows的idea开发，我们一般只能通过docker容器-p暴露的端口去连接kafka，所以每个kafka容器还需要设置一个KAFKA_ADVERTISED_LISTENERS来告诉宿主机的客户端，如何连接到自己，这里需要使用localhost+暴露在宿主机的端口。

那么如果KAFKA_ADVERTISED_LISTENERS里面有2个地址，如何保证broker之间的连接使用的是容器名，而宿主机客户端使用的是localhost呢？

这需要KAFKA_INTER_BROKER_LISTENER_NAME来指定前者。

并且由于KAFKA_ADVERTISED_LISTENERS里面有2个地址，所以我们还需要KAFKA_LISTENER_SECURITY_PROTOCOL_MAP来映射监听器名字。

3. Docker Compose

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
version: '3.8'

services:
  zookeeper:
    image: bitnami/zookeeper:3.8.2
    container_name: zookeeper
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
    networks:
      - kafka

  kafka1:
    image: bitnami/kafka:3.6.1
    container_name: kafka1
    depends_on:
      - zookeeper
    ports:
      - "19092:9092"
    environment:
      - KAFKA_BROKER_ID=1
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=INTERNAL://0.0.0.0:9093,EXTERNAL://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka1:9093,EXTERNAL://localhost:19092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
    networks:
      - kafka

  kafka2:
    image: bitnami/kafka:3.6.1
    container_name: kafka2
    depends_on:
      - zookeeper
    ports:
      - "29092:9092"
    environment:
      - KAFKA_BROKER_ID=2
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=INTERNAL://0.0.0.0:9093,EXTERNAL://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka2:9093,EXTERNAL://localhost:29092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
    networks:
      - kafka

  kafka3:
    image: bitnami/kafka:3.6.1
    container_name: kafka3
    depends_on:
      - zookeeper
    ports:
      - "39092:9092"
    environment:
      - KAFKA_BROKER_ID=3
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_LISTENERS=INTERNAL://0.0.0.0:9093,EXTERNAL://0.0.0.0:9092
      - KAFKA_ADVERTISED_LISTENERS=INTERNAL://kafka3:9093,EXTERNAL://localhost:39092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL
    networks:
      - kafka

networks:
  kafka:
    driver: bridge

可以看到每个容器都设置了INTERNAL，因为指定了KAFKA_INTER_BROKER_LISTENER_NAME=INTERNAL，所以这是用于broker之间的连接，其监听在本地的0.0.0.0:9093，广播给其它broker的通信地址是<容器名>:9093，使用PLAINTEXT（明文）方式通信。

除此之外还设置了EXTERNAL，监听在本地的0.0.0.0:9092，广播给客户端的地址是localhost:19092、localhost:29092、localhost:39092，也就是windows上的客户端通过localhost:19092访问broker，这会被docker的-p映射到对应容器的9092，被0.0.0.0:9092对接。

4. 验证

连接到某个容器。创建test主题。

1
kafka-topics.sh --create --topic test --partitions 3 --replication-factor 3 --bootstrap-server kafka1:9093

查看分区和副本情况，可以看到在不同的broker上，输出中显示的是Broker ID。

1
2
3
4
5
I have no name!@7212060b6e3d:/$ kafka-topics.sh --describe --topic test --bootstrap-server kafka1:9093
Topic: test     TopicId: Lo1eQ6aCQj6WiFcNiVBrcw PartitionCount: 3       ReplicationFactor: 3    Configs: 
        Topic: test     Partition: 0    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
        Topic: test     Partition: 1    Leader: 3       Replicas: 3,1,2 Isr: 3,1,2
        Topic: test     Partition: 2    Leader: 1       Replicas: 1,2,3 Isr: 1,2,3

引入pom包

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
<dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>3.6.1</version>
        </dependency>
        <!-- SLF4J API -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.36</version>
        </dependency>
        <!-- Logback classic (SLF4J implementation) -->
        <dependency>
            <groupId>ch.qos.logback</groupId>
            <artifactId>logback-classic</artifactId>
            <version>1.2.11</version>
        </dependency>

生产者代码，通过localhost:19092连接到集群。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
package org.dragon.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.HashMap;

public class KafkaProducerTest {
    public static void main(String[] args) throws InterruptedException {
        //创建producer
        HashMap<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config);

        for (int i = 0; i < 10; i++) {
            //创建record
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                    "test",
                    "key"+i,
                    "我是你爹"+i
            );
            //发送record
            producer.send(record);
            Thread.sleep(500);
        }

        //关闭producer
        producer.close();
    }
}

消费者代码，

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
package org.dragon.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Collections;
import java.util.HashMap;

public class KafkaConsumerTest {
    public static void main(String[] args) {

        // 创建消费者对象
        HashMap<String, Object> config = new HashMap<>();
        config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
        config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        config.put(ConsumerConfig.GROUP_ID_CONFIG, "mkl");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(config);
        // 消费者订阅主题
        kafkaConsumer.subscribe(Collections.singletonList("test"));

        try {
            while (true){
                // 消费者拉取消息
                ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                records.forEach(System.out::println);
            }
        }finally {
            // 消费者关闭
            kafkaConsumer.close();
        }
    }
}

都启动后，消费者和生产者日志正常。

Kafka基础组件图推演
发表更新SpringCloud / Kafka5 分钟读完 (大约685个字)

Kafka基础组件图推演

1. Controller Broker

Kafka集群中有一个Controller Broker，负责元数据管理和协调。

Kafka使用Zookeeper作为集群元数据的存储和管理工具。Zookeeper保存了集群的状态信息，包括所有的Topic、分区、Leader和副本信息等。

当集群状态发生变化时，Controller Broker会将变更信息写入Zookeeper。

当现任Controller Broker发生故障时，Kafka集群中的其他Broker会检测到这一情况，并通过Zookeeper进行选举。

一个Broker成功竞选为新的Controller Broker后，会从Zookeeper读取最新的集群元数据。

保障机制

  • 持久化元数据：Zookeeper持久化存储集群的元数据，确保在任何时刻都可以获取到最新的状态信息。
  • 心跳机制：Broker与Zookeeper之间通过心跳机制保持连接，快速检测Controller Broker的故障。
  • 选举机制：通过Zookeeper的选举机制，能够快速选举出新的Controller Broker，并从Zookeeper同步元数据。

2. 组件架构

1. Log Manager

LogManager主要负责管理Kafka日志（log）的存储和检索。

比如：生产者将消息发送到Partition0的Leader Broker1。LogManager在Broker1上将消息写入Partition0的日志文件。

2. Replication Manager

ReplicationManager主要负责管理分区数据的复制和同步。

每个分区的Leader和Follower之间的同步是独立进行的。也就是说，每个分区都有自己的同步过程，不依赖于其他分区。

虽然每个分区的同步过程是独立的，但每个Broker会为它所管理的每个分区（无论是Leader还是Follower）启动相应的复制线程，这些线程负责处理具体的同步任务。

比如：ReplicationManager在Broker1上将新写入的消息推送到Partition0的Follower Broker2和Broker3。ReplicationManager在Broker2和Broker3上处理从Broker1接收的复制请求，将消息写入它们各自的日志文件。

3. SocketServer

SocketServer是Kafka Broker中的一个组件，负责处理网络连接和I/O操作。它负责接受来自客户端和其他Broker的连接请求，并为每个连接分配一个线程进行处理。

4. NetworkServer

NetworkServer是Kafka的网络通信框架的一个核心部分，负责管理和调度网络请求。它使用了NIO（非阻塞I/O）来处理高并发的网络连接。

5. ZKClient

与Zookeeper通信的组件。

Kafka基础框架图推演
发表更新SpringCloud / Kafka9 分钟读完 (大约1328个字)

Kafka基础框架图推演

1. 单点模型

1. 名词概念

  1. Broker：是指Kafka集群中的一个节点。一个Kafka集群由多个Broker组成，这些Broker共同协作来处理消息的存储、传输和消费。Broker管理一个或多个分区。

  2. Topic：生产者将消息发送到指定的Topic，消费者订阅Topic以获取消息。Topic本身只是一个逻辑上的分组，不具有物理存储的概念。

  3. Partition：是Topic的子集，是Kafka中实际存储和处理消息的基本单元。每个Topic可以分成多个Partition，每个Partition都是一个有序的、不可变的消息序列。

  4. Replica：分区可以存在多副本。

  5. Leader Broker：分区的多副本下，负责处理所有的读写请求的分区所处的broker。

  6. FollowerBroker：分区的多副本下，负责同步Leader的数据的分区所处的broker。

生产者把消息（record）发送到kafka，消费者通过偏移量（offset，类似数组的下标）获取数据。

同时每个分区会有自己的Log文件，kafka使用log文件来把数据保存到磁盘。

2. 分布式集群-横向扩展

1. Topic多分区

关于生产

生产者通过Bootstrap Broker连接到Kafka集群。这一步是为了建立初始连接，并获取集群的元数据。

一旦生产者获取了这些元数据，它就知道每个分区的Leader Broker是谁，从而可以将消息直接发送到正确的Leader Broker。

生产者发送消息时必须指定Topic，但是分区是可选的。

  • 不指定分区：如果生产者没有手动指定分区，Kafka会根据默认的分区策略将消息分配到分区。默认的分区策略如下：
    • 如果消息有键（Key），Kafka会根据键的哈希值来确定分区。相同的键总是被分配到同一个分区。
    • 如果消息没有键（Key），Kafka会使用轮询或随机的方式将消息分配到分区，以确保消息分布均匀。
  • 指定分区：生产者也可以在发送消息时明确指定分区。这样，消息会直接发送到指定的分区。

在Kafka中，生产者将消息发送到Broker时，Broker的第一个操作就是将消息记录到磁盘中，以确保消息的持久性和可靠性。

关于消费

Kafka中的消费者通常属于一个消费者组（Consumer Group）。每个消费者组有一个唯一的组ID。消费者组的概念用于实现消息的负载均衡和并行消费。

当多个消费者属于同一个组时，Kafka会将Topic的分区分配给组内的消费者。**每个分区只能由组内的一个消费者消费**，这样可以实现负载均衡。

  • 单个消费者订阅一个Topic：

    • 如果只有一个消费者订阅了一个Topic，那么该消费者会接收到该Topic中的所有消息。

  • 多个消费者属于同一个组：

    • Topic中的分区会在组内的消费者之间进行分配。每个分区只会被组内的一个消费者消费。
    • 如果消费者数量多于分区数，多余的消费者将不会分配到任何分区，处于空闲状态。这些消费者可以在有其他消费者退出时自动接管其分区，从而实现高可用性。
    • 如果消费者数量少于分区数，有些消费者会被分配多个分区。

  • 多个消费者属于不同的组：

    • 每个组都会独立消费Topic中的所有消息。也就是说，消息会被广播到所有组中的消费者。

关于分区新增

Kafka会在集群中创建新的分区。这些新的分区会被分配到不同的Broker，以实现数据的均衡存储和高可用性。Kafka不会自动将现有分区的数据重新分配或均衡到新的分区。新的分区从创建时开始是空的，只有在后续生产者发送消息时，才会向这些新的分区写入数据。消费者组会感知到分区数量的变化，并触发重新平衡。

2. 分区多副本

Kafka允许每个分区有多个副本（Replica），这些副本存储在不同的Broker上。一个副本被称为Leader，负责处理所有的读写请求，其他副本为Follower，负责同步Leader的数据。

多个副本同时只有一个副本可以读写，这就是Leader副本，其他副本成为Follower副本，用作备份。

docker部署简单的Kafka
发表更新SpringCloud / Kafka4 分钟读完 (大约595个字)

docker部署简单的Kafka

1. 拉取镜像

选择一组兼容性好的版本。

1
2
docker pull bitnami/kafka:3.6.1
docker pull bitnami/zookeeper:3.8.2

2. 运行

创建网络

首先，创建一个名为 kafka 的 Docker bridge 网络：

1
docker network create kafka

运行 ZooKeeper 容器

然后，运行 ZooKeeper 容器并将其连接到 kafka 网络：

1
docker run -d --name zookeeper --network kafka -e ALLOW_ANONYMOUS_LOGIN=yes bitnami/zookeeper:3.8.2

运行 Kafka 容器

最后，运行 Kafka 容器并将其连接到 kafka 网络：

1
docker run -d --name kafka --network kafka -e KAFKA_BROKER_ID=1 -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 -e ALLOW_PLAINTEXT_LISTENER=yes -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 -p 9092:9092 bitnami/kafka:3.6.1

这些命令将使 ZooKeeper 和 Kafka 容器在同一个 Docker 网络中运行，并确保它们可以相互通信。

3. 简单的校验

要判断 ZooKeeper 和 Kafka 容器是否正常运行，可以通过以下几个步骤进行检查：

1. 检查容器状态

首先，检查 ZooKeeper 和 Kafka 容器是否正在运行：

1
docker ps

输出应包含类似以下内容：

1
2
3
CONTAINER ID   IMAGE                        COMMAND                  CREATED          STATUS          PORTS                                        NAMES
<zookeeper_id> bitnami/zookeeper:3.8.2      "/opt/bitnami/script…"   <some_time_ago>  Up <some_time>  2181/tcp                                     zookeeper
<kafka_id>     bitnami/kafka:3.6.1          "/opt/bitnami/script…"   <some_time_ago>  Up <some_time>  0.0.0.0:9092->9092/tcp, :::9092->9092/tcp   kafka

2. 检查 ZooKeeper 日志

查看 ZooKeeper 容器的日志，以确保它已成功启动并正在运行：

1
docker logs zookeeper

日志中应包含类似以下内容：

1
2
INFO  Started AdminServer on address 0.0.0.0, port 8080
INFO  binding to port 0.0.0.0/0.0.0.0:2181

3. 检查 Kafka 日志

查看 Kafka 容器的日志，以确保它已成功连接到 ZooKeeper 并正在运行：

1
docker logs kafka

日志中应包含类似以下内容：

1
2
INFO  [KafkaServer id=1] started (kafka.server.KafkaServer)
INFO  [ZooKeeperClient] Connected. (org.apache.zookeeper.ClientCnxn)

4. 使用 Kafka 命令行工具检查

进入 Kafka 容器内部，并使用 Kafka 命令行工具检查 Kafka 和 ZooKeeper 的状态：

1
2
3
4
docker exec -it kafka /bin/bash

# 列出 Kafka 主题
kafka-topics.sh --list --bootstrap-server kafka:9092

如果可以成功列出 Kafka 主题，则表示 Kafka 和 ZooKeeper 正常运行。

5. 创建和删除测试主题

可以尝试创建一个测试主题，并查看是否成功：

1
2
3
4
5
6
7
8
# 创建一个名为 test-topic 的主题
kafka-topics.sh --create --topic test-topic --partitions 1 --replication-factor 1 --bootstrap-server kafka:9092

# 列出所有主题，确认 test-topic 是否存在
kafka-topics.sh --list --bootstrap-server kafka:9092

# 删除 test-topic 主题
kafka-topics.sh --delete --topic test-topic --bootstrap-server kafka:9092

通过以上步骤，可以确认 ZooKeeper 和 Kafka 容器是否正常运行并相互通信。

Hexo结合多个主题扩展为Gallery画廊并实现文章加密
发表更新Hexo5 分钟读完 (大约813个字)

Hexo结合多个主题扩展为Gallery画廊并实现文章加密

可能参考的文章：

如何优雅的使用Github Action服务来将Hexo部署到Github Pages - Hexo

当前PC环境中有Node和Git。版本可以参考Hexo文档。

文章中出现的yourusername为Github用户名，your-repo为仓库名。

1. 初始化

新建一个项目根目录HexoNote，并初始化hexo（主Hexo）。然后新建一个子文件夹hexoB，在里面也初始化Hexo（副Hexo），用于存放第二个hexo。副Hexo的主题是一个简单的主题，很适合作为画廊。

1
2
3
4
5
6
根目录
    - public
    - hexoB
        - public
        - ...
    - ...

把项目传达github，私有仓库，名为HexoNotePrivate。

假设存放Hexo编译后的静态文件夹的仓库名为HexoNote，公共的public。

我们利用github action编译HexoNotePrivate里面的2个hexo，然后整合静态文件夹，推送到HexoNote仓库，HexoNote配置github pages。

2. 安装加密

1
npm install --save hexo-blog-encrypt

在主Hexo的_config.yml文件新增配置

1
2
3
4
5
6
7
8
# Security
encrypt: # hexo-blog-encrypt
  abstract: 这是一篇加密文章，需要密码才能继续阅读。
  message: 当前文章暂不对外可见，请输入密码后查看！
  tags:
  - {name: private, password: MKL1009}
  wrong_pass_message: 抱歉，您输入的密码错误，请检查后重新输入。
  wrong_hash_message: 抱歉, 当前文章不能被校验, 不过您还是可以看看解密后的内容。

在md中的tags中新增private即可设置为加密文章。

1
2
tags:
  - private

3. 配置文件

把主Hexo的_config.yml文件修改

1
2
url: https://yourusername.github.io/HexoNote
root: /HexoNote/

并且在主Hexo的主题中新增菜单（导航栏）的一个元素，相对地址为/gallery，具体按主题的文档设置

1
2
3
4
5
6
7
navbar:
    # Navigation menu items
    menu:
        Home: /
        Archives: /archives
        Gallery: /gallery
        About: /about

把副Hexo的_config.yml文件修改

1
2
url: https://yourusername.github.io/HexoNote/gallery
root: /HexoNote/gallery/

4. 创建Token

详情看顶部的文章链接，把token放入private私有仓库。

5. 新建公开仓库

新建公开仓库HexoNote，作为github pages的代码仓库。

6. 工作流

新建文件.github\workflows\deploy.yml

node版本和推送地址要按情况修改。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
name: Deploy Hexo to GitHub Pages

on:
  push:
    branches:
      - main  # 当推送到 main 分支时触发

jobs:
  build:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout repository
        uses: actions/checkout@v2
        with:
          submodules: false  # 禁用子模块检查

      - name: Setup Node.js
        uses: actions/setup-node@v2
        with:
          node-version: '18'

      - name: Install Dependencies
        run: npm install

      - name: Install Hexo Git Deployer
        run: |
          npm install hexo-deployer-git --save
          npm install hexo-cli -g

      - name: Clean and Generate Static Files for Theme A
        run: |
          hexo clean
          hexo generate

      - name: Generate Static Files for Theme B
        run: |
          cd hexoB
          npm install
          npm install hexo-deployer-git --save
          npm install hexo-cli -g
          hexo clean
          hexo generate
          cd ..

      - name: Move Theme B Output to Gallery
        run: |
          mv hexoB/public public/gallery

      - name: Configure Git
        run: |
          git config --global user.name 'github-actions[bot]'
          git config --global user.email 'github-actions[bot]@users.noreply.github.com'

      - name: Deploy to GitHub Pages
        env:
          GH_TOKEN: ${{ secrets.GH_TOKEN }}
        run: |
          cd public/
          git init
          git add -A
          git commit -m "Create by workflows"
          git remote add origin https://${{ secrets.GH_TOKEN }}@github.com/yourusername/HexoNote.git
          git push origin HEAD:gh-pages -f


7. 实现效果

1. 加密

2. 画廊B主题

Typora + Hexo 图片路径问题(Typedown)
发表更新Hexo3 分钟读完 (大约430个字)

Typora + Hexo 图片路径问题(Typedown)

1. 冲突来源

Hexo上对于图片在md中的引用，使用了post_asset_folder: true配置，来更好的管理图片。
当一篇名为xxx.md的文章引用1.png图片时，默认让1.png保持在xxx文件夹下，那么md中即可使用{% asset_img 1.png %}来引用图片。

而typora中，或者Typedown中，复制图片时，一般使用![](./xxx/1.png)。

2. 解决思路

  1. 让每次图片复制到md时，typora都能将其自动放入和md文件同名同级文件夹下。
  2. 然后在Hexo编译前使用脚本将![](./xxx/1.png)转化为{% asset_img 1.png %}，并且保持md源文件不变。

3. 实现

1. typora图片路径

这很简单。

但是如果你是typedown就会发现，不支持解析${filename}，那么只有每次写的时候手动选择同级同名文件夹了。

2. hexo脚本

在scripts\before_generate.js中写入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
// const path = require('path');

// hexo.extend.filter.register('before_post_render', data => {
//   if (data.layout === 'post') {
//     const postName = path.basename(data.source, '.md');
//     const imgRegex = new RegExp(`!\\[.*?\\]\\(\\.\\/${postName}\\/([^\\)]+)\\)`, 'g');

//     data.content = data.content.replace(imgRegex, (match, p1) => {
//       return `{% asset_img ${p1} %}`;
//     });
//   }
//   return data;
// });

const path = require('path');

hexo.extend.filter.register('before_post_render', data => {
  if (data.layout === 'post') {
    const postName = path.basename(data.source, '.md');
    const imgRegex = new RegExp(`!\\[.*?\\]\\(\\.\\/${postName}\\/([^\\)]+)\\)`, 'g');

    // 原始内容
    const originalContent = data.content;

    // 转换内容
    let match;
    let modifiedContent = originalContent;
    while ((match = imgRegex.exec(originalContent)) !== null) {
      const originalLine = match[0];
      const newLine = `{% asset_img ${match[1]} %}`;
      
      // 打印转换前后的对比
      console.log(`Original line: ${originalLine}`);
      console.log(`Converted line: ${newLine}\n`);

      // 进行替换
      modifiedContent = modifiedContent.replace(originalLine, newLine);
    }

    // 更新数据内容
    data.content = modifiedContent;
  }
  return data;
});

被注释掉了是不会打印日志对比前后修改的，没注释的会。

执行hexo clean和hexo generate，然后hexo server看看效果。

Nginx基础概念和常用操作
发表更新Front-End / Nginx26 分钟读完 (大约3826个字)

Nginx基础概念和常用操作

1. 安装、启动、连接

直接拿docker安装，拉取镜像，运行，开放端口。

1
docker pull nginx:stable-perl

虽然在容器中性能有所损失，但是方便，加快学习进度。

启动后浏览器访问localhost，写着Welcome等欢迎语，就是启动成功了。

在VSCode下载一些Docker常用插件，然后就可以拿VSCode直接连上’正在运行的容器’，和连接Linux服务器一样。

输入nginx -V，查看nginx信息，包括安装目录、编译参数、配置文件位置、日志文件位置等信息。可以看到配置文件位于--conf-path=/etc/nginx/nginx.conf 。

1
2
3
4
5
6
root@f4f6c922d837:/# nginx -V
nginx version: nginx/1.26.1
built by gcc 12.2.0 (Debian 12.2.0-14) 
built with OpenSSL 3.0.11 19 Sep 2023
TLS SNI support enabled
configure arguments: --prefix=/etc/nginx --sbin-path=/usr/sbin/nginx --modules-path=/usr/lib/nginx/modules --conf-path=/etc/nginx/nginx.conf --error-log-path=/var/log/nginx/error.log --http-log-path=/var/log/nginx/access.log --pid-path=/var/run/nginx.pid --lock-path=/var/run/nginx.lock --http-client-body-temp-path=/var/cache/nginx/client_temp --http-proxy-temp-path=/var/cache/nginx/proxy_temp --http-fastcgi-temp-path=/var/cache/nginx/fastcgi_temp --http-uwsgi-temp-path=/var/cache/nginx/uwsgi_temp --http-scgi-temp-path=/var/cache/nginx/scgi_temp --user=nginx --group=nginx --with-compat --with-file-aio --with-threads --with-http_addition_module --with-http_auth_request_module --with-http_dav_module --with-http_flv_module --with-http_gunzip_module --with-http_gzip_static_module --with-http_mp4_module --with-http_random_index_module --with-http_realip_module --with-http_secure_link_module --with-http_slice_module --with-http_ssl_module --with-http_stub_status_module --with-http_sub_module --with-http_v2_module --with-http_v3_module --with-mail --with-mail_ssl_module --with-stream --with-stream_realip_module --with-stream_ssl_module --with-stream_ssl_preread_module --with-cc-opt='-g -O2 -ffile-prefix-map=/data/builder/debuild/nginx-1.26.1/debian/debuild-base/nginx-1.26.1=. -fstack-protector-strong -Wformat -Werror=format-security -Wp,-D_FORTIFY_SOURCE=2 -fPIC' --with-ld-opt='-Wl,-z,relro -Wl,-z,now -Wl,--as-needed -pie'

2. 快速尝试部署网站

查看nginx配置文件，可以找到如下部分。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
server {
    listen       80;
    listen  [::]:80;
    server_name  localhost;

    location / {
        root   /usr/share/nginx/html;
        index  index.html index.htm;
    }
    error_page   500 502 503 504  /50x.html;
    location = /50x.html {
        root   /usr/share/nginx/html;
    }
}

这意味着，nginx 把所有对服务器根路径的请求，代理到 /usr/share/nginx/html 目录下，并且把 index.html 或 index.htm 作为默认页面。

如果你使用Vue或者Hex开发网站，它们的打包输出目录，大致结构分别如下

1
2
3
4
5
6
7
dist/
├── css/
│   └── app.12345.css
├── js/
│   └── app.12345.js
├── index.html
└── favicon.ico
1
2
3
4
5
6
7
8
9
10
public/
├── css/
│   └── style.css
├── js/
│   └── script.js
├── index.html
├── about/
│   └── index.html
└── archives/
    └── index.html

你只需要把dist或者public下的所有文件，直接复制到/usr/share/nginx/html目录下，重启nginx即可。

3. 配置文件

这个版本的nginx自带2个配置文件，首先是nginx.conf。

如果你修改了配置文件，可以用nginx -t来校验配置是否合法。然后使用nginx -s reload来重新加载配置文件。

1. nginx.conf

全局配置

1
2
3
4
5
user  nginx;
worker_processes  auto;

error_log  /var/log/nginx/error.log notice;
pid        /var/run/nginx.pid;

  1. user 指令:

    • 用法：user <username>;
    • 描述：指定 Nginx 运行时使用的系统用户。
    • 示例：user nginx; 表示使用 nginx 用户。

  2. worker_processes 指令:

    • 用法：worker_processes <number|auto>;
    • 描述：设置工作进程数，auto 会自动根据 CPU 核心数设置。
    • 示例：worker_processes 4; 表示使用 4 个工作进程。

  3. error_log 指令:

    • 用法：error_log <file> <level>;
    • 描述：设置错误日志的路径和日志级别。
    • 日志级别选项：
      • debug：调试信息
      • info：一般信息
      • notice：通知信息（默认）
      • warn：警告信息
      • error：错误信息
      • crit：严重错误信息
      • alert：需要立即处理的问题
      • emerg：紧急情况
    • 示例：error_log /var/log/nginx/error.log notice; 表示记录通知级别及以上的日志。

  4. pid 指令:

    • 用法：pid <file>;
    • 描述：指定存储 Nginx 进程 ID 的文件路径。
    • 示例：pid /var/run/nginx.pid; 表示将 PID 存储在 /var/run/nginx.pid 文件中。

可能改动的部分：

  1. worker_processes auto;

    • 自动设置为 CPU 核心数，适用于大多数情况。

    • 可以手动设置为特定的进程数以微调性能，但一般不需要。

事件模块

1
2
3
events {
    worker_connections  1024;
}
  1. worker_connections 指令:
    • 用法：worker_connections <number>;
    • 描述：设置每个工作进程允许的最大连接数。
    • 示例：worker_connections 1024; 表示每个工作进程允许最多 1024 个连接。

可能改动的部分：

  1. worker_connections 1024;

    • 默认设置为 1024，适用于大多数小到中型网站。

    • 对于高流量网站，可以增加连接数，以提高并发处理能力。

    • 需要根据系统的 ulimit 设置进行调整，确保系统支持更多的连接。

HTTP 模块

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
http {
    include       /etc/nginx/mime.types;
    default_type  application/octet-stream;

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

    access_log  /var/log/nginx/access.log  main;

    sendfile        on;
    #tcp_nopush     on;

    keepalive_timeout  65;

    #gzip  on;

    include /etc/nginx/conf.d/*.conf;
}

  1. include 指令:

    • 用法：include <file|directory>;
    • 描述：包含指定的文件或目录下的所有配置文件。
    • 示例：include /etc/nginx/mime.types; 包含 MIME 类型配置文件。

  2. default_type 指令:

    • 用法：default_type <MIME-type>;
    • 描述：设置默认的 MIME 类型。
    • 示例：default_type application/octet-stream; 表示未能确定文件类型时使用 application/octet-stream。

  3. log_format 指令:

    • 用法：log_format <name> <format>;

    • 描述：定义日志格式。

    • 示例：

      1
      2
      3
      log_format main '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

      定义名为 main 的日志格式。

  4. access_log 指令:

    • 用法：access_log <file> <format>;
    • 描述：设置访问日志的路径和日志格式。
    • 示例：access_log /var/log/nginx/access.log main; 使用 main 格式记录访问日志。

  5. sendfile 指令:

    • 用法：sendfile <on|off>;
    • 描述：启用或禁用 sendfile 选项，用于提高文件传输效率。
    • 示例：sendfile on; 启用 sendfile。

  6. tcp_nopush 指令:

    • 用法：tcp_nopush <on|off>;
    • 描述：用于优化传输数据时的 TCP 性能，通常与 sendfile 一起使用。
    • 示例：#tcp_nopush on; 默认注释掉。

  7. keepalive_timeout 指令:

    • 用法：keepalive_timeout <timeout>;
    • 描述：设置保持客户端连接的超时时间，单位是秒。
    • 示例：keepalive_timeout 65; 设置超时时间为 65 秒。

  8. gzip 指令:

    • 用法：gzip <on|off>;
    • 描述：启用或禁用 gzip 压缩。
    • 示例：#gzip on; 默认注释掉。

  9. include指令

    • 用法：include <file|directory>;
    • 描述：include 指令用于包含其他配置文件或目录中的所有配置文件。这可以帮助将配置文件分割成更小的部分，以便于管理和维护。
    • 示例：
      include /etc/nginx/mime.types;：包含 MIME 类型配置文件。
      include /etc/nginx/conf.d/*.conf;：包含 /etc/nginx/conf.d/ 目录下的所有以 .conf 结尾的配置文件。

可能改动的部分：

  1. keepalive_timeout 65;

    • 默认设置为 65 秒，适用于大多数情况。

    • 对于高并发的场景，可以适当调低以减少资源占用。

  2. tcp_nopush on;

    • 通常注释掉，如果启用，可以优化数据传输，尤其是在发送大文件时。

    • 启用 tcp_nopush 后，Nginx 会尝试将数据块一次性发送到网络，而不是逐个包地发送，从而减少包的数量，提高传输效率。

    • 通常与sendfile on;一起使用。启用 sendfile 后，数据在内核空间中直接从一个文件描述符传输到另一个文件描述符，避免了在用户空间和内核空间之间的多次拷贝，从而提高了传输效率。

  3. gzip on;

    • 通常开启后，可以启用 gzip 压缩以减少传输数据量，提高传输速度。

    • 启用后还需配置 gzip_types 和其他参数，以确保正确压缩所需的文件类型。如

      1
      2
      3
      4
      http {
          gzip on;
          gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript;
      }

    • 其他 gzip 配置选项：

      • gzip_min_length：设置压缩的最小文件大小。
      • gzip_comp_level：设置压缩级别（1-9），级别越高压缩率越大，但消耗的 CPU 资源更多。
      • gzip_buffers：设置用于存储压缩结果的缓冲区大小。
      • gzip_vary：设置 Vary: Accept-Encoding 响应头，指示代理服务器和浏览器缓存可以根据请求的 Accept-Encoding 头进行不同的缓存。

    • Gzip 压缩是通过压缩算法（如 DEFLATE）将 HTTP 响应内容压缩成更小的体积，然后再发送给客户端。解压缩的过程是nginx和浏览器自动完成的。

性能优化建议

  • **增加 worker_connections**：

    • 提高每个工作进程的最大连接数可以显著提升 Nginx 的并发处理能力。
    • 需要确保系统的文件描述符限制足够高。

  • 启用 gzip 压缩：

    • 减少传输数据量，尤其适用于文本类型的资源（如 HTML、CSS、JavaScript）。

    • 配置示例：
      nginx
      复制代码
      gzip on; gzip_types text/plain text/css application/json application/javascript text/xml application/xml application/xml+rss text/javascript;

  • **优化 keepalive_timeout**：

    • 在高并发情况下，适当调低 keepalive 超时时间以减少长时间占用连接资源。

  • **使用 tcp_nopush 和 tcp_nodelay**：

    • 对于需要优化数据传输性能的场景，可以启用 tcp_nopush 和 tcp_nodelay。

2. default.conf

好的，让我们对 default.conf 文件的每一行进行解析：

server 块

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
server {
    listen       80;
    listen  [::]:80;
    server_name  localhost;

    #access_log  /var/log/nginx/host.access.log  main;

    location / {
        root   /usr/share/nginx/html;
        index  index.html index.htm;
    }

    #error_page  404              /404.html;

    # redirect server error pages to the static page /50x.html
    #
    error_page   500 502 503 504  /50x.html;
    location = /50x.html {
        root   /usr/share/nginx/html;
    }
}

基本设置

  1. **server { ... }**：

    • 定义一个虚拟服务器块，包含服务器的配置。

  2. **listen 80;**：

    • 监听 IPv4 的 80 端口，用于处理 HTTP 请求。

  3. **listen [::]:80;**：

    • 监听 IPv6 的 80 端口，用于处理 HTTP 请求。

  4. **server_name localhost;**：

    • 定义服务器的名称为 localhost，用于匹配请求的 Host 头。

日志设置

  1. **#access_log /var/log/nginx/host.access.log main;**：
    • 配置访问日志的路径和格式，此处被注释掉。如果启用，将使用 main 日志格式记录访问日志。

根路径设置

  1. **location / { ... }**：

    • 配置根路径（即所有请求）的处理方式。

  2. **root /usr/share/nginx/html;**：

    • 指定请求的根目录为 /usr/share/nginx/html。

  3. **index index.html index.htm;**：

    • 指定默认的索引文件为 index.html 或 index.htm。

4. 反向代理

正向代理是代理客户端，客户端对于服务端来说是不可见的。

反向代理是代理服务端，服务端对于客户端来说是不可见的。

1. 模拟3个Web

1
docker pull python:3.9-slim

在桌面或任意地方新建app.py，内容为

1
2
3
4
5
6
7
8
9
10
11
12
from flask import Flask
import os

app = Flask(__name__)

@app.route('/')
def hello():
    return f"Hello from the backend server running on port {os.environ.get('PORT')}!"

if __name__ == "__main__":
    app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)))

启动3个python容器，模拟3个web服务。

1
2
3
4
5
6
7
8
C:\Users\mumu\Desktop>docker run --name web1 -d -e PORT=8081 -p 8081:8081 -v /c/Users/mumu/Desktop/app.py:/app/app.py python:3.9-slim sh -c "pip install flask && python /app/app.py"
ab2ded77eafa82fbf4027f5714e5cb71bcb9775696f0e8c5423a70a7916a80ac

C:\Users\mumu\Desktop>docker run --name web2 -d -e PORT=8082 -p 8082:8082 -v /c/Users/mumu/Desktop/app.py:/app/app.py python:3.9-slim sh -c "pip install flask && python /app/app.py"
e6c5321ea3926157dc2f2d9ed38df53c715da51c496c5ff0e7ff0c4e68c85696

C:\Users\mumu\Desktop>docker run --name web3 -d -e PORT=8083 -p 8083:8083 -v /c/Users/mumu/Desktop/app.py:/app/app.py python:3.9-slim sh -c "pip install flask && python /app/app.py"
f0bea8ac956af515fe35159fe94c86e919dfd566b782d06641472621a6299e26

需要保证nginx和3个python在同一个docker网络中，默认应该都是bridge。

2. 链接

使用docker inspect bridge查看刚才3个容器的ip。然后修改nginx的default.conf文件。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
upstream backend {
    server 172.17.0.3:8081;
    server 172.17.0.4:8082;
    server 172.17.0.5:8083;
}

server {
    listen       80;
    listen  [::]:80;
    server_name  localhost;

    location / {
        root   /usr/share/nginx/html;
        index  index.html index.htm;
    }

    location /app {
        proxy_pass http://backend/;
    }

    error_page   500 502 503 504  /50x.html;
    location = /50x.html {
        root   /usr/share/nginx/html;
    }
}

  • **upstream backend**：定义了一个名为 backend 的上游服务器组。

  • **location /app { ... }**：配置如何处理以 /app 开头的所有请求。

  • **proxy_pass http://backend/;**：将这些请求代理到上游服务器组 backend。请求的路径 /app 的前缀将被去除，然后传递给上游服务器。例如，/app/foo 将被代理为 /foo。

重启服务。

访问localhost/app并多次刷新试试。

5. 负载均衡

在不断刷新页面会发现，反向代理是以“轮询”的方式，将请求分发到3个web服务之一的。

除了轮询之外，还有以下方式。

1. 加权轮询，Weighted Round Robin

可以为不同的服务器设置权重，权重越高，服务器被请求的概率越大。

1
2
3
4
5
upstream backend {
    server 172.17.0.3:8081 weight=3;
    server 172.17.0.4:8082 weight=2;
    server 172.17.0.5:8083 weight=1;
}

weight默认为1 。

2. 最少连接，Least Connections

将请求分发给当前连接数最少的服务器，适用于处理时间长、连接占用多的请求。

1
2
3
4
5
6
upstream backend {
    least_conn;
    server 172.17.0.3:8081;
    server 172.17.0.4:8082;
    server 172.17.0.5:8083;
}

3. IP哈希，IP Hash

根据客户端 IP 地址进行哈希计算，保证同一客户端的请求总是转发到同一台服务器，适用于会话保持。

1
2
3
4
5
6
upstream backend {
    ip_hash;
    server 172.17.0.3:8081;
    server 172.17.0.4:8082;
    server 172.17.0.5:8083;
}

4. 哈希，Hash

根据自定义的键值进行哈希计算来分发请求。

1
2
3
4
5
6
upstream backend {
    hash $request_uri;
    server 172.17.0.3:8081;
    server 172.17.0.4:8082;
    server 172.17.0.5:8083;
}

可以使用以下变量作为键值：

  • $remote_addr：客户端 IP 地址
  • $request_uri：请求的 URI
  • $http_cookie：请求的 Cookie

ip_hash和hash $remote_addr一样都可以实现同一个客户端的请求分配到同一个后端服务器。但是一般这种情况使用前者更多。

6. HTTPS

首先要安装有openssl，然后创建一个存放证书和私钥的文件夹，比如mkdir C:\nginx\ssl

执行

1
openssl req -x509 -nodes -days 365 -newkey rsa:2048 -keyout C:\nginx\ssl\nginx-selfsigned.key -out C:\nginx\ssl\nginx-selfsigned.crt

按问题填写信息，就可以。

得到2个文件后，复制到容器中，比如/etc/nginx/ssl目录下，然后再default.conf中添加一个server块，如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
server {
    listen 443 ssl;
    listen [::]:443 ssl;
    server_name localhost;

    ssl_certificate /etc/nginx/ssl/nginx-selfsigned.crt;
    ssl_certificate_key /etc/nginx/ssl/nginx-selfsigned.key;

    location / {
        root   /usr/share/nginx/html;
        index  index.html index.htm;
    }

    location /app {
        proxy_pass http://backend/;
    }

    error_page  500 502 503 504  /50x.html;
    location = /50x.html {
        root   /usr/share/nginx/html;
    }
}

打开浏览器访问https://localhost

7. 虚拟主机

一个server块就是一个虚拟主机。

新建一个conf文件和default.conf同一目录。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
upstream backend2 {
    server 172.17.0.3:8081 weight=2;
    server 172.17.0.4:8082 weight=2;
    server 172.17.0.5:8083 weight=6;
}

server {
    listen       81;
    listen  [::]:81;
    server_name  localhost;

    location / {
        root   /usr/share/nginx/html;
        index  index.html index.htm;
    }

    location /app {
        proxy_pass http://backend2/;
    }

    error_page   500 502 503 504  /50x.html;
    location = /50x.html {
        root   /usr/share/nginx/html;
    }
}

重启服务，浏览器访问http://localhost:81/app 。

利用这种方式，我们可以在同一台机器上部署多个网站。

关注我

分类

标签

Consumer1
Hexo3
JWT3
Kafka15
Nginx1
OAuth2.12
RSA1
Spring Security2
SpringBoot1
Vue31
ambari2
c++1
css32
docker29
echarts1
ecmascript1
elasticsearch1
excel3
gateway3
git1
github1
hadoop2
hdfs1
hexo1
intellij-idea1
java25
javascript9
jenkins1
jvm1
kafka1
linux2
md1
mybatis5
mysql4
nginx1
openresty1
postman1
rabbitmq6
redis11
restful1
ruby2
ruoyi1
spring5
spring boot21
spring cloud4
spring security1
springamqp1
springboot7
springcloud3
typescript2
ubuntu1
vue6
vue.js8
web1
web安全1
windows1
word1
zookeeper1
个人博客2
事务1
分区1
分布式4
分布式文件系统1
前端12
单例模式1
博客1
双检锁1
反向代理1
后端12
大数据1
存储1
容器10
幂等1
开发语言3
微服务3
拦截器1
排序算法1
支付1
支付宝1
数据库5
文件系统1
服务器4
未分类1
枚举单例1
架构10
正则表达式1
消费者1
爬虫1
生产者7
缓存10
网站2
视频1
设计模式1
负载均衡1
运维3
重构1
锁1
集群1

最新文章

归档

×