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");});
});
)
