Nodejs 第七十六章（MQ进阶）

MQ介绍和基本使用在上一章介绍过了，不再重复

消息：在RabbitMQ中，消息是传递的基本单元。它由消息体和可选的属性组成
生产者Producer：生产者是消息的发送方，它将消息发送到RabbitMQ的交换器（Exchange）中
交换器Exchange：交换器接收从生产者发送的消息，并根据特定的规则将消息路由到一个或多个队列中
队列Queue：队列是消息的接收方，它存储了待处理的消息。消费者可以从队列中获取消息并进行处理
消费者Consumer：消费者是消息的接收方，它从队列中获取消息并进行处理

MQ进阶用法

发布订阅

发布订阅，消息的发送者称为发布者（Publisher），而接收消息的一个或多个实体称为订阅者（Subscriber）

回顾上一篇，点对点通讯生产者发送一条消息通过路由投递到Queue，只有一个消费者能消费到也就是一对一发送

请添加图片描述

回归主题发布订阅就是生产者的消息通过交换机写到多个队列，不同的订阅者消费不同的队列，也就是实现了一对多

发布订阅的模式分为四种

Direct（直连）模式：把消息放到交换机指定key的队列里面。
Topic（主题）模式：把消息放到交换机指定key的队列里面，额外增加使用"*“匹配一个单词或使用”#"匹配多个单词
Headers（头部）模式：把消息放到交换机头部属性去匹配队列
Fanout（广播）模式：把消息放入交换机所有的队列，实现广播

发布订阅-代码编写

1. direct模式编写

主要就是通过 routingKey 匹配实现路由这里的zs就是routingKey

生产者

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
//创建一个频道
const channel = await connection.createChannel() 
//声明一个交换机
/*** @param {String} exchange 交换机的名称* @param {String} type "direct" | "topic" | "headers" | "fanout" | "match" | 使用广播模式* @param {Object} options {durable: true} //开启消息持久化*/
await channel.assertExchange('logs', 'direct', {durable: true
})
//发送消息
/*** @param {String} exchange 交换机的名称* @param {String} routingKey 路由键* @param {Buffer} content 消息内容*///这里的zs就是routingKey
channel.publish('logs', 'zs', Buffer.from('小满direct模式发送的消息'))//断开
await channel.close()
await connection.close()
process.exit(0)

消费者(我们编写多个方便测试)

consume.js

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
const channel = await connection.createChannel() //创建一个频道await channel.assertExchange('logs', 'direct', {durable: true
})//添加一个队列
const { queue } = await channel.assertQueue('queue1', {durable: true
})
//绑定交换机
/*** @param {String} queue 队列名称* @param {String} exchange 交换机名称* @param {String} routingKey 路由键*/
//匹配对应的zs值才能收到
await channel.bindQueue(queue, 'logs', 'zs')
//接收消息
channel.consume('queue1', (msg) => {console.log(msg.content.toString());
}, {noAck: true //自动确认消息被消费
})

consume2.js

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
const channel = await connection.createChannel() //创建一个频道await channel.assertExchange('logs', 'direct', {durable: true
})//添加一个队列
const { queue } = await channel.assertQueue('queue2', {durable: true
})
//绑定交换机
/*** @param {String} queue 队列名称* @param {String} exchange 交换机名称* @param {String} routingKey 路由键*///匹配对应的zs值才能收到
await channel.bindQueue(queue, 'logs', 'zs')
//接收消息
channel.consume('queue2', (msg) => {console.log(msg.content.toString());
}, {noAck: true //自动确认消息被消费
})

2. Topic模式编写

我们把模式切换成了Topic 并且publish 发布的时候 routingKey 换成了 xm.xxxxxxxx

生产者

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
//创建一个频道
const channel = await connection.createChannel() 
//声明一个交换机
/*** @param {String} exchange 交换机的名称* @param {String} type "direct" | "topic" | "headers" | "fanout" | "match" | 使用广播模式* @param {Object} options {durable: true} //开启消息持久化*/
await channel.assertExchange('topic', 'topic', {durable: true
})
//发送消息
/*** @param {String} exchange 交换机的名称* @param {String} routingKey 路由键* @param {Buffer} content 消息内容*///注意这儿匹配规则换了 换成xm.xxxxxxxxxxxxxxxxxxxxx
channel.publish('logs', 'xm.sadsdsdasdasdasdsda', Buffer.from('小满topic模式发送的消息'))//断开
await channel.close()
await connection.close()
process.exit(0)

消费者匹配(注意这里匹配规则xm.*'使用了* 就是模糊匹配的意思)

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
const channel = await connection.createChannel() //创建一个频道await channel.assertExchange('topic', 'topic', {durable: true
})//添加一个队列
const { queue } = await channel.assertQueue('queue1', {durable: true
})
//绑定交换机
/*** @param {String} queue 队列名称* @param {String} exchange 交换机名称* @param {String} routingKey 路由键 *匹配一个单词 #匹配多个单词*///这儿变化了
await channel.bindQueue(queue, 'topic', 'xm.*')
//接收消息
channel.consume('queue1', (msg) => {console.log(msg.content.toString());
}, {noAck: true //自动确认消息被消费
})

3. Headers模式

生产者(注意 publish 增加第四个参数开启了header 添加了data参数)

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
//创建一个频道
const channel = await connection.createChannel() //声明一个交换机/*** @param {String} exchange 交换机的名称* @param {String} type "direct" | "topic" | "headers" | "fanout" | "match" | 使用广播模式* @param {Object} options {durable: true} //开启消息持久化*/await channel.assertExchange('headers', 'headers', {durable: true})//发送消息/*** @param {String} exchange 交换机的名称* @param {String} routingKey 路由键* @param {Buffer} content 消息内容* @param {Object} options {headers: {'key': 'value'}} //定义匹配规则*///嘿 这儿变了channel.publish('headers', '', Buffer.from('小满headers模式发送的消息'),{headers: {data:'xmzs'}})//断开await channel.close()await connection.close()process.exit(0)

消费者(bindQueue 增加一个对象属性跟生产者对应即可)

   import amqplib from 'amqplib'const connection = await amqplib.connect('amqp://localhost:5672')const channel = await connection.createChannel() //创建一个频道await channel.assertExchange('headers', 'headers')//添加一个队列const { queue } = await channel.assertQueue('queue1')//绑定交换机/*** @param {String} queue 队列名称* @param {String} exchange 交换机名称* @param {String} routingKey 路由键 *匹配一个单词 #匹配多个单词*/await channel.bindQueue(queue, 'headers', '',{data:'xmzs' //注意这儿不加headers 直接放值即可})//接收消息channel.consume(queue, (msg) => {console.log(msg.content.toString());}, {noAck: true //自动确认消息被消费})

4. Fanout模式

生产者(其实也就是routingKey 变成一个空值实现全体广播)

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
//创建一个频道
const channel = await connection.createChannel()
//声明一个交换机
/**
* @param {String} exchange 交换机的名称
* @param {String} type "direct" | "topic" | "headers" | "fanout" | "match" | 使用广播模式
* @param {Object} options {durable: true} //开启消息持久化
*/
await channel.assertExchange('fanout', 'fanout')
//发送消息
/**
* @param {String} exchange 交换机的名称
* @param {String} routingKey 路由键
* @param {Buffer} content 消息内容
*/
channel.publish('fanout', '', Buffer.from('小满fanout模式发送的消息'))//断开
await channel.close()
await connection.close()
process.exit(0)

消费者(routingKey接受空值即可就算有值也会被忽略)

import amqplib from 'amqplib'
const connection = await amqplib.connect('amqp://localhost:5672')
const channel = await connection.createChannel() //创建一个频道
await channel.assertExchange('fanout', 'fanout')//添加一个队列
const { queue } = await channel.assertQueue('queue1')
//绑定交换机
/**
* @param {String} queue 队列名称
* @param {String} exchange 交换机名称
* @param {String} routingKey 路由键 *匹配一个单词 #匹配多个单词
*/
await channel.bindQueue(queue, 'fanout', '')
//接收消息
channel.consume(queue, (msg) => {
console.log(msg.content.toString());
}, {
noAck: true //自动确认消息被消费
})