RabbitMQ 的分片插件（rabbitmq_sharding）允許將消息分布到多個隊列中，這在消息量很大或處理速度要求高的情況下非常有用。分片功能通過將消息拆分到多個隊列中來平衡負載，從而提升消息處理的吞吐量和可靠性。它能夠在多個隊列之間分配負載，避免單個隊列過載。（注：不能單獨消費分片消息。消息分片不利于消息順序區分）

啟用消息分片插件。?

rabbitmq-plugins enable rabbitmq_sharding 

示例

通過rabbitmq management添加策略，用于分片消息匹配轉發。

或者通過命令添加策略?

CTL set_policy images-shard "queue10" '{"shards-per-node": 3, "routing-key": "sharding"}'

producer.ts

import RabbitMQ from 'amqplib';async function start() {try {const conn = await RabbitMQ.connect("amqp://admin:admin1234@localhost:5672//mirror?heartbeat=60");conn.on("error", function (err1) {if (err1.message !== "Connection closing") {console.error("[AMQP] conn error", err1.message);}});conn.on("close", function () {console.error("[AMQP] reconnecting");return setTimeout(start, 1000);});console.log("[AMQP] connected");let channel = null;try {channel = await conn.createChannel();} catch (err) {console.error("[AMQP]", err);return setTimeout(start, 1000);}const exchangeName = 'exchange_queue10';await channel.assertExchange(exchangeName,'x-modulus-hash',{durable: true,arguments: {'x-modulus': 3 // 分片數量（需與隊列分片數匹配）}},);let routeKey = '';for (let i = 0; i < 1000; ++i) {// console.log('message send!', channel.sendToQueue(//   queueName,//   Buffer.from(`發送消息,${i}${Math.ceil(Math.random() * 100000)}`),//   { persistent: true, correlationId: 'ooooooooooooooo' },// 消息持久化，重啟后存在//   // (err: any, ok: Replies.Empty)=>{}// ));let num = Math.ceil(Math.random() * 100000);console.log('消息發送是否成功', num, routeKey, channel.publish(exchangeName,`${routeKey}${i}`,Buffer.from(`"發送消息, index:${i}, number:${num}, routeKey:${JSON.stringify(routeKey)}"`),{persistent: true,},));}setTimeout(() => {conn.close();process.exit(0);}, 1000);} catch (err) {console.error("[AMQP]", err);return setTimeout(start, 1000);}
}start();

consumer.ts

import RabbitMQ, { type Replies } from 'amqplib/callback_api';RabbitMQ.connect('amqp://admin:admin1234@localhost:5672//mirror', (err0, conn) => {if (err0) {console.error(err0);return;}conn.createChannel(function (err1, channel) {console.log('[*] waiting...');const exchangeName = 'exchange_queue10';channel.prefetch(32);// for(let i=0;i<3;++i){//   channel.assertQueue(queueName, { durable: true }, () => {//     channel.bindQueue(queueName, exchangeName, `shard_${shardId}`);//   });// }channel.consume(exchangeName, function (msg) {if(msg){console.log(`隊列'${exchangeName}'接收到的消息`, msg?.content.toString());// 第二個參數，false拒絕當前消息// 第二個參數，true拒絕小于等于當前消息// 第三個參數，3false從隊列中清除// 第三個參數，4true從新在隊列中排隊channel.nack(msg, false, false);}}, {// noAck: true, // 是否自動確認消息，為true不需要調用channel.ack(msg);noAck: false,arguments: {}}, (err: any, ok: Replies.Empty) => {console.log(err, ok);});});conn.on("error", function (err1) {if (err1.message !== "Connection closing") {console.error("[AMQP] conn error", err1.message);}});conn.on("close", function () {console.error("[AMQP] reconnecting");});
});