RabbitMq-SuperStream

  • Carter Ho
  • 13 Minutes
  • 2026年10月12日

甚麼是SuperStream

要了解SuperStream前須要了解Stream，RMQ在3.9的時候推出了Stream；在此之前RMQ依循著AMQP 0.9.1讓exchange, route, queue照著協定處理訊息，並且擁有ack機制來表示訊息是否被處理過

在一些情境，過往的訊息可能會被需要提出重新檢視，原先的ack機制就無法達成這個需求，所以有了Stream這個功能；Stream可以把訊息持久話(但儲存一定有上限，所以還是要訂定保留策略)，讓過去的訊息可以被提出重新檢視；但持久化勢必會影響讀取效率，所以Stream透過二進制的方式提供服務來達成最佳吞吐量

Stream特性

SuperStream在3.11時加入，是基於Stream的情境下對Queue實行partition策略，讓原子性的應用可以更彈性

Classic Queue概觀

Stream Queue概觀

Stream Queue概觀

SuperStream叢集會用到的Port

運行SuperStream啟用差件

1
2
3
4
5
6
7
8
9
10
## 需要四個插件啟用: rabbitmq_management,rabbitmq_stream,rabbitmq_stream_management,rabbitmq_consistent_hash_exchange

## 方法一  透過rabbitmq cli啟用
rabbitmq rabbitmq-plugins enable {插件名稱}

## 方法二  因為方法一透過容器啟用都會失敗 所以直接透過config更改
pwd: etc/rabbitmq/rabbitmq.conf
content:

[rabbitmq_management,rabbitmq_stream,rabbitmq_stream_management,rabbitmq_consistent_hash_exchange].

運行SuperStream Host設定

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
## 方法一 透過環境變數賦予
RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS='-rabbitmq_stream advertised_host "{宿主的IP/HOST}" -rabbitmq_stream advertised_port {宿主監聽的PORT}'

## 方法二 因為方法一透過容器啟用都會失敗  所以直接透過config更改
pwd: /etc/rabbitmq/rabbitmq.conf
content: 

stream.advertised_host = {實體IP/HOST}
stream.advertised_port = {super stream暴露在宿主身上的Port}

## 若有叢集可以順便設定
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_classic_config
cluster_formation.classic_config.nodes.1 = rabbit@{host-1}
cluster_formation.classic_config.nodes.2 = rabbit@{host-2}
cluster_formation.classic_config.nodes.3 = rabbit@{host-3}

SuperStream配置partition跟HA policy

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
## 方法一  透過rabbitmq cli指令操作
rabbitmq rabbitmq-streams add_super_stream {SuperStream名稱} --partitions {partition數量}
rabbitmq rabbitmqctl set_policy stream-ha "^{SuperStream名稱}" '{"queue-leader-locator":"balanced"}' --apply-to queues

## 方法二  透過程式建立

var topic = "topic";
var subscribeTopic = "subscribeTopic";
var partitionCount = 3;
var exchangeName = $"{topic}-{subscribeTopic}";

var exchangeArgs = new Dictionary<string, object>()
{
    { "x-super-stream", true },
    { "hash-header", "PartitionKey" }
};

// 綁定SuperStream Exchange
await channel.ExchangeDeclareAsync(
    exchange: exchangeName,
    type: "x-consistent-hash",
    durable: true,
    autoDelete: false,
    arguments: exchangeArgs);

var queueArgs = new Dictionary<string, object>()
{
    // 表示Stream
    { "x-queue-type", "stream" },
    // 保留 1 天 (單位支援時間字串，如 "1D" 或以毫秒/秒為單位)
    { "x-max-age", "1D" },
    // 每個 Segment 最大 500MB (Bytes)
    { "x-stream-max-segment-size-bytes", 524_288_000 }
	// HA策略
	{ "x-initial-cluster-size", rmqClusterNum }
};

// 建立partition
for (var partition = 0; partition < partitionCount; partition++)
{
    var queueName = $"{exchangeName}-{partition}";

    // 建立Q
    await channel.QueueDeclareAsync(
        queue: queueName,
        durable: true,
        exclusive: false,
        autoDelete: false,
        arguments: queueArgs);

    // 建立Q跟Exchange關係
    await channel.QueueBindAsync(
        queue: queueName,
        exchange: exchangeName,
        routingKey: "100");
}

OffsetSpec的類型

OffsetSpec.OffsetTypeNext ConsumerConfig必要設定

1
2
3
ConsumerConfig.Reference = "{唯一識別ID}"
ConsumerConfig.Reference = true
ConsumerConfig.Reference = Task<(string reference, stream stream, bool isActive)>

參考資料

範例程式

© 2026 Carter`s Blog