尚硅谷rabbitmq2024 集群篇流式队列第54节答疑

二、负载均衡
在文件/etc/haproxy/haproxy.cfg末尾追加:
frontend rabbitmq_stream_frontendbind 192.168.200.100:33333mode tcp
default_backend rabbitmg_stream_backend
backend rabbitmg_stream_backendmode tcp
balance roundrobin
server rabbitmg1 192.168.200.100:555 checkserver rabbitmg2 192.168.200.150:5551 check
server rabbitmg3 192.168.200.200:5552 check
----为什么启用stream队列之后，端口变成了5552？

在RabbitMQ中，启用Stream队列是一种新的队列类型，专门用于处理高吞吐量和持久化消息的场景。Stream队列的实现和普通队列有所不同，特别是在网络和端口配置上。

### 为什么启用Stream队列后端口变成了5552？

启用Stream队列后，RabbitMQ可能会使用不同的端口来处理Stream协议的通信。这是因为Stream队列和普通队列在实现和优化上有不同的需求。以下是一些可能的原因：

1. **协议差异**：
- Stream队列使用的是RabbitMQ Stream协议，这与AMQP协议有所不同。为了区分和优化不同协议的通信，RabbitMQ可能会使用不同的端口。

2. **性能优化**：
- 为了优化Stream队列的性能，RabbitMQ可能会将Stream协议的通信分配到专用的端口上，以避免与普通AMQP通信混淆，从而提高传输效率和可靠性。

3. **配置灵活性**：
- 使用不同的端口可以提供更大的配置灵活性。例如，可以为Stream队列的通信分配更多的网络资源或应用特定的网络策略。

### 配置示例

在HAProxy配置文件中，你可以根据实际情况配置不同的端口。例如：

```haproxy
frontend rabbitmq_stream_frontend
bind 192.168.200.100:33333
mode tcp
default_backend rabbitmg_stream_backend

backend rabbitmg_stream_backend
mode tcp
balance roundrobin
server rabbitmg1 192.168.200.100:555 check
server rabbitmg2 192.168.200.150:5551 check
server rabbitmg3 192.168.200.200:5552 check
```

在这个示例中，HAProxy会将流量分发到不同的RabbitMQ节点上，并且每个节点可能会使用不同的端口来处理Stream协议的通信。

### 确认端口配置

要确认RabbitMQ Stream队列使用的端口，可以检查RabbitMQ的配置文件（如`rabbitmq.conf`）或使用RabbitMQ管理插件查看节点的监听端口配置。

### RabbitMQ配置文件示例

假设你在`rabbitmq.conf`中配置了Stream队列的端口：

```ini
listeners.tcp.default = 5672
listeners.stream.default = 5552
```

在这个示例中，默认的AMQP通信使用端口5672，而Stream协议的通信使用端口5552。

### 总结

启用Stream队列后，RabbitMQ使用不同的端口（如5552）来处理Stream协议的通信，这是为了区分不同协议的通信、优化性能以及提供配置灵活性。你可以在HAProxy配置文件中相应地配置这些端口，以确保负载均衡正常工作。如果有其他问题或需要进一步
========
就是说新开了一个端口，原来的端口不变？

是的，当启用 RabbitMQ 的 Stream 队列时，通常会新开一个端口用于处理 Stream 协议的通信，而原来的 AMQP 通信端口保持不变。这种做法有助于区分不同类型的通信，优化性能，并提供更大的配置灵活性。

### 详细解释

1. **新开端口**：
- 启用 Stream 队列后，RabbitMQ 会使用一个新的专用端口来处理 Stream 协议的通信。例如，在前面的示例中，Stream 协议的通信使用端口 `5552`。

2. **原有端口不变**：
- 原有的 AMQP 通信端口保持不变。例如，默认的 AMQP 通信端口通常是 `5672`。这个端口继续用于处理传统的 AMQP 协议通信。

### 配置示例

假设你在 `rabbitmq.conf` 中配置了不同的端口：

```ini
listeners.tcp.default = 5672
listeners.stream.default = 5552
```

在这个配置中：
- `5672` 端口用于处理传统的 AMQP 协议通信。
- `5552` 端口用于处理 Stream 协议通信。

### 在 HAProxy 中的配置

在 HAProxy 的配置文件中，你可以根据实际情况配置不同的端口。例如：

```haproxy
frontend rabbitmq_stream_frontend
bind 192.168.200.100:33333
mode tcp
default_backend rabbitmg_stream_backend

backend rabbitmg_stream_backend
mode tcp
balance roundrobin
server rabbitmg1 192.168.200.100:5552 check
server rabbitmg2 192.168.200.150:5552 check
server rabbitmg3 192.168.200.200:5552 check
```

在这个示例中，HAProxy 会将流量分发到不同的 RabbitMQ 节点上，并且每个节点使用 `5552` 端口来处理 Stream 协议的通信。

### 小结

- 启用 Stream 队列后，RabbitMQ 会新开一个端口（如 `5552`）用于处理 Stream 协议的通信。
- 原来的 AMQP 通信端口（如 `5672`）保持不变。
- 这种配置方式有助于区分不同类型的通信，优化性能，并提供更大的配置灵活性。

如果你有更多问题或需要进一步的帮助，请随时告诉我！

============
backend rabbitmg_stream_backendmode tcp
balance roundrobin
server rabbitmg1 192.168.200.100:5552 checkserver rabbitmg2 192.168.200.150:5552 checkserver rabbitmg3 192.168.200.200:5552 check
-----check是干什么的

在HAProxy配置中，`check`关键字用于启用健康检查功能。健康检查的目的是定期检测后端服务器的健康状态，如果某个服务器不可用（例如宕机或响应不正常），HAProxy会自动将其从负载均衡池中移除，直到该服务器恢复正常。

### 具体作用

1. **定期检查**：
- HAProxy会定期向配置的后端服务器发送健康检查请求，以确保服务器处于可用状态。

2. **自动移除不可用服务器**：
- 如果健康检查失败（服务器没有响应或响应不正常），HAProxy会自动将该服务器从负载均衡池中移除，避免将流量分发到不可用的服务器。

3. **自动恢复**：
- 一旦服务器恢复正常并通过健康检查，HAProxy会自动将其重新加入负载均衡池。

### 配置示例

在你的HAProxy配置中，`check`关键字的使用如下：

```haproxy
backend rabbitmg_stream_backend
mode tcp
balance roundrobin
server rabbitmg1 192.168.200.100:5552 check
server rabbitmg2 192.168.200.150:5552 check
server rabbitmg3 192.168.200.200:5552 check
```

### 解释

- `server rabbitmg1 192.168.200.100:5552 check`：这行配置表示HAProxy会定期向`192.168.200.100:5552`发送健康检查请求，以确保该服务器处于健康状态。
- `check`关键字启用健康检查功能。

### 健康检查的类型

健康检查可以有不同的类型，具体取决于你的需求：
- **TCP检查**：默认情况下，HAProxy会进行简单的TCP连接检查，确保服务器能够接受TCP连接。
- **HTTP检查**：如果你的服务是HTTP/HTTPS，可以配置HTTP健康检查，通过发送HTTP请求并检查响应状态码来判断服务器是否健康。

### HTTP检查示例

如果你需要进行HTTP健康检查，可以这样配置：

```hapro

=============
<dependency>
<groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.30</version></dependency>
<dependency>
<groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>1.2.3</version>I</dependency>
----日志相关的包为什么有两个？

在Java项目中，日志记录是非常重要的功能，而`SLF4J`和`Logback`是常用的日志记录工具。你提到的两个依赖项分别是`SLF4J`和`Logback`，它们在日志记录中扮演不同的角色。下面是对这两个依赖项的详细解释：

### 1. SLF4J (Simple Logging Facade for Java)

#### 作用：
- **日志门面**：SLF4J是一个日志门面（Facade），它提供了统一的日志记录接口，但不负责实际的日志记录实现。通过SLF4J，你可以在代码中使用统一的日志记录API，而不必关心底层的日志实现。
- **灵活性**：使用SLF4J可以让你在不同的日志实现之间切换而无需修改代码。例如，你可以在开发阶段使用Logback，在生产环境中使用Log4j，只需修改配置而无需更改代码。

#### 依赖项：
```xml
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.30</version>
</dependency>
```

### 2. Logback

#### 作用：
- **日志实现**：Logback是一个具体的日志实现框架，负责实际的日志记录操作。它是由SLF4J的作者开发的，作为Log4j的继任者，提供了更高效、灵活和强大的日志记录功能。
- **配置灵活**：Logback支持多种配置方式（如XML、Groovy），可以根据不同的需求进行灵活配置。

#### 依赖项：
```xml
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>1.2.3</version>
</dependency>
```

### 为什么需要两个依赖？

1. **抽象与实现分离**：
- `SLF4J`提供了统一的日志记录API，是抽象层。

============
log4j被淘汰了吗

Log4j 并没有被完全淘汰，但其使用确实有所下降，尤其是在某些场景中被替代。以下是一些关键点，可以帮助你理解为什么 Log4j 的使用在某些情况下减少了，以及它的现状。

### Log4j 的现状

1. **Log4j 2 的推出**：
- Log4j 1.x 系列已经停止维护，最后一个版本是 1.2.17。为了应对 Log4j 1.x 的一些局限性和安全问题，Apache 推出了 Log4j 2，它是一个全新的、改进的日志框架，提供了更高的性能和更多的功能。

2. **安全问题**：
- Log4j 1.x 存在一些已知的安全漏洞，这使得很多开发者转向了更安全的日志框架，如 Log4j 2 或 Logback。

3. **性能和功能**：
- Log4j 2 和 Logback 都提供了比 Log4j 1.x 更高的性能和更多的功能。Logback 是由 SLF4J 的作者设计的，作为 Log4j 的继任者，具有更好的配置灵活性和性能优化。

### Log4j 2 的优势

- **异步日志记录**：通过异步日志记录，可以显著提高性能，尤其是在高吞吐量的应用中。
- **插件架构**：Log4j 2 采用了插件架构，增加了扩展的灵活性。
- **更好的配置选项**：支持 XML、JSON、YAML 等多种配置格式。
- **API 兼容性**：提供了对 Log4j 1.x API 的兼容支持，方便从 Log4j 1.x 迁移到 Log4j 2。

### 为什么选择 Logback 和 SLF4J

- **统一的日志接口**：SLF4J 提供了一个统一的日志接口，可以在不同的日志实现之间轻松切换。
- **灵活的配置**：Logback 提供了非常灵活的配置选项，支持多种格式（如 XML 和 Groovy）。
- **性能优化**：Logback 经过高度优化，适用于高性能应用。
- **社区支持**：Logback 和 SLF4J 拥有活跃的社区和良好的文档支持。

### 总结

Log4j 并没有被完全淘汰，但在某些场景中，其使用确实减少了，特别是 Log4j 1.x 系列。为了更好的性能、安全性和功能，很多开发者选择了 Logback 或升级到 Log4j 2。SLF4J 提供了一个统一的日志接口，使得在不同日志实现之间切换变得更加容易。

如果你正在使用 Log4j 1.x，强烈建议升级到 Log4j 2 或切换到 Logback，以确保你的应用程序具备更好的性能和安全性。

=============
[2]解析
在 Environment 中封装的连接信息仅负责连接到broker
Producer 在构建对象时会访问 broker 拉取集群中 Leader 的连接信息
。将来实际访问的是集群中的 Leader 节点
Leader 的连接信息格式是:节点名称:口号
Environment environment =Environment.builder().host("192.168.20日.10日”)
.port(33333)
.username("atguigu")
.password("123456")
连接
broker
.build();
Producer producer=environment.producerBuilder().stream("stream.atguigu.test")
访问
.build();
发布消息(
node02:5552
Leader的连接信息
-----为什么要找node02,代码里不是写了100吗？

你的问题涉及到 RabbitMQ 的 Stream 队列和其内部的 Leader 节点机制。让我详细解释一下为什么代码中指定的连接信息和实际访问的 Leader 节点信息可能会有所不同。

### 背景

在 RabbitMQ 中，Stream 队列是一种高吞吐量、持久化的队列类型。为了实现高可用性和负载均衡，Stream 队列通常会在集群中分布，并且每个队列都有一个 Leader 节点和多个 Follower 节点。

### 连接信息和 Leader 节点

1. **初始连接信息**：
- `Environment` 中封装的连接信息（如 `host` 和 `port`）用于初始连接到 RabbitMQ broker。这只是一个入口点，用于连接到集群中的某个节点。

2. **拉取 Leader 信息**：
- 一旦连接到 broker 后，Producer 会从 broker 拉取 Stream 队列的元数据，包括该队列的 Leader 节点信息。
- 这个 Leader 节点是实际负责处理该 Stream 队列请求的节点。

### 示例代码解释

```java
Environment environment = Environment.builder()
.host("192.168.200.100")
.port(33333)
.username("atguigu")
.password("123456")
.build();

Producer producer = environment.producerBuilder()
.stream("stream.atguigu.test")
.build();
```

在这段代码中：

- `Environment` 中的 `host` 和 `port`（`192.168.200.100:33333`）用于初始连接到 RabbitMQ 集群中的某个节点。
- 连接到这个节点后，Producer 会请求 Stream 队列 `stream.atguigu.test` 的元数据。

### 为什么实际访问的是 `node02:5552`？

- **Leader 节点选择**：
- Stream 队列在集群中有多个副本，其中一个是 Leader 节点。Producer 从初始连接的节点获取到 `stream.atguigu.test` 队列的 Leader 信息。
- 这个 Leader 节点可能是 `node02`，并且它监听的端口是 `5552`。

- **动态分配**：
- RabbitMQ 的 Stream 队列的 Leader 节点是动态分配的，并不一定是初始连接的节点。因此，尽管你最初连接的是 `192.168.200.100:33333`，实际处理请求的可能是 `node02:5552`。

### 小结

- 初始连接信息（`192.168.200.100:33333`）用于连接到 RabbitMQ 集群中的某个节点。
- Producer 会从这个节点拉取 Stream 队列的 Leader 信息。
- 实际处理 Stream 队列请求的 Leader 节点可能是集群中的其他节点（如 `node02:5552`）。

希望这能帮助你理解为什么代码中指定的连接信息和实际访问的 Leader 节点信息会有所不同。如果有其他疑问，随时问我！你的问题涉及到 RabbitMQ 的 Stream 队列和其内部的 Leader 节点机制。让我详细解释一下为什么代码中指定的连接信息和实际访问的 Leader 节点信息可能会有所不同。

### 背景

### 连接信息和 Leader 节点

### 示例代码解释

```java
Environment environment = Environment.builder()
.host("192.168.200.100")
.port(33333)
.username("atguigu")
.password("123456")
.build();

Producer producer = environment.producerBuilder()
.stream("stream.atguigu.test")
.build();
```

在这段代码中：

### 为什么实际访问的是 `node02:5552`？

### 小结

希望这能帮助你理解为什么代码中指定的连接信息和实际访问的 Leader 节点信息会有所不同。如果有其他疑问，随时问我！

==============
CountDownLatch countDownLatchnewCountDownLatch(1):
producer.send(
producer.messageBuilder().addData(messagePayload).build()
confirmationStatus>if(confirmationStatus.isConfirmed()){System.out.println("[生产者端]the message made it to the broker");
}else{System.out.println("[生产者端]the message did not make it to the broker");
}):
countDownLatch.countDown();
countDownLatch.await();
----countDownLatch干嘛的？

`CountDownLatch` 是 Java 中一个非常有用的同步工具类，位于 `java.util.concurrent` 包中。它允许一个或多个线程等待，直到在其他线程中执行的一组操作完成。

### 主要功能

`CountDownLatch` 通过一个计数器来实现。计数器的初始值由创建 `CountDownLatch` 对象时指定。每当一个线程完成某个操作后，调用 `countDown()` 方法，计数器的值就会减 1。当计数器的值达到 0 时，所有在 `await()` 方法上等待的线程都会被唤醒，继续执行。

### 使用场景

- **确保某些操作完成后再继续执行**：例如，在主线程中等待多个工作线程完成各自的任务。
- **并行计算的结果汇总**：等待所有子任务完成，然后汇总结果。

### 代码示例解释

```java
CountDownLatch countDownLatch = new CountDownLatch(1);

producer.send(
producer.messageBuilder().addData(messagePayload).build(),
confirmationStatus -> {
if (confirmationStatus.isConfirmed()) {
System.out.println("[生产者端]the message made it to the broker");
} else {
System.out.println("[生产者端]the message did not make it to the broker");
}
countDownLatch.countDown();
}
);

countDownLatch.await();
```

#### 解释

1. **创建 `CountDownLatch` 对象**：
```java
CountDownLatch countDownLatch = new CountDownLatch(1);
```
这里创建了一个 `CountDownLatch` 对象，并将计数器初始化为 1。

2. **发送消息**：
```java
producer.send(
producer.messageBuilder().addData(messagePayload).build(),
confirmationStatus -> {
if (confirmationStatus.isConfirmed()) {
System.out.println("[生产者端]the message made it to the broker");
} else {
System.out.println("[生产者端]the message did not make it to the broker");
}
countDownLatch.countDown();
}
);
```
这里发送了一条消息，并注册了一个回调函数。当消息发送完成后，回调函数会被调用。回调函数中：
- 检查 `confirmationStatus` 是否确认消息已经到达 broker。
- 打印相应的消息。
- 调用 `countDownLatch.countDown()`，将计数器减 1。

3. **等待**：
```java
countDownLatch.await();
```
主线程调用 `countDownLatch.await()`，在计数器变为 0 之前会一直阻塞。因为计数器初始值为 1，只有当回调函数中调用 `countDownLatch.countDown()` 后，计数器变为 0，主线程才会继续执行。

### 总结

`CountDownLatch` 在这个示例中的作用是：

- 确保主线程等待消息发送完成（并且收到确认或未确认的回调）后再继续执行。
- 使用 `countDownLatch.await()` 方法让主线程阻塞，直到消息发送回调函数调用 `countDownLatch.countDown()` 方法，计数器变为 0。

通过这种方式，可以确保在消息发送完成并得到确认或未确认的结果后，主线程才会继续执行后续操作。

==============