1
2
3
4
5
6
7
8
9

Broker:它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保证数据能按照指定的方式进行传输,
Exchange：消息交换机,它指定消息按什么规则,路由到哪个队列。
Queue:消息的载体,每个消息都会被投到一个或多个队列。
Binding:绑定，它的作用就是把exchange和queue按照路由规则绑定起来.
Routing Key:路由关键字,exchange根据这个关键字进行消息投递。
vhost:虚拟主机,一个broker里可以有多个vhost，用作不同用户的权限分离。
Producer:消息生产者,就是投递消息的程序.
Consumer:消息消费者,就是接受消息的程序.
Channel:消息通道,在客户端的每个连接里,可建立多个channel.

$ node -v
v8.9.3
# 为了执行ts代码
$ ts-node -v
v9.0.0
$ brew install rabbitmq
$ brew services start rabbitmq

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

// emit_log_direct.ts 生产者
import * as amqp from 'amqplib'const url = `amqp://localhost`;(async function publish(){const exchange = 'direct_logs';const msg = 'hello world';const routingKeys = ['info', 'error', 'warning'];// 1. 创建链接const connect = await amqp.connect(url);// 2. 创建channelconst channel = await connect.createChannel();// 3. 创建or连上 交换机// 3.1 直连方式；await channel.assertExchange(exchange, 'direct', { durable: false });let i = 0;while(i<1){const index = random(3);// 4. 消息发给交换机channel.publish(exchange, routingKeys[index], Buffer.from(msg));console.log(`[x] Sent ${msg}-- ${routingKeys[index]}`);i++;}await sleep(1);await connect.close();process.exit(0);
})();function sleep(time: number) {return new Promise((resolve) => setTimeout(resolve, time*1000));
}function random(max: number){return Math.floor(Math.random() * Math.floor(max));
}

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

// receive_logs_direct.ts 消费者
import * as amqp from 'amqplib'
const url = `amqp://localhost`;(async function receive(){const connect = await amqp.connect(url);const channel = await connect.createChannel();const exchange = 'direct_logs';await channel.assertExchange(exchange, 'direct', { durable: false });const routingKeys = ['info', 'error', 'warning'];const queueA = await channel.assertQueue('queueA');await channel.bindQueue(queueA.queue, exchange, routingKeys[0]);await channel.bindQueue(queueA.queue, exchange, routingKeys[1]);await channel.bindQueue(queueA.queue, exchange, routingKeys[2]);const queueB = await channel.assertQueue('queueB');await channel.bindQueue(queueB.queue, exchange, routingKeys[1]);channel.consume(queueA.queue, msg => {console.log("队列AAAAA：", msg.content.toString())}, { noAck: true });channel.consume(queueB.queue, msg => {console.log("队列BBBBBBBB：", msg.content.toString())}, { noAck: true });// await connect.close();
})();

1
2
3
4
5
6
7
8
9
10

# 执行
$ ts-node emit_log_direct.ts
[x] Sent hello world-- error
[x] Sent hello world-- info$ ts-node receive_logs_direct.ts
队列AAAAA： hello world
队列BBBBBBBB： hello world队列AAAAA： hello world

1
2

* (star) : 匹配一个单词
# (hash) : 匹配 0 或 更多单词

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

// emit_log_topic.ts
import * as amqp from 'amqplib'const url = `amqp://localhost`;(async function publish(){const exchange = 'topic_logs';const msg = 'hello world';// bothqueue、firstqueue、secondqueueconst keys = ['quick.orange.rabbit', 'quick.orange.fox', 'lazy.brown.fox'];// 1. 创建链接const connect = await amqp.connect(url);// 2. 创建channelconst channel = await connect.createChannel();// 3. 创建or连上 交换机// 3.1 直连方式；await channel.assertExchange(exchange, 'topic', { durable: false });let i = 0;while(i<1){const index = random(3);// 4. 消息发给交换机channel.publish(exchange, keys[index], Buffer.from(msg));console.log(`[x] Sent ${msg}-- ${keys[index]}`);i++;}await sleep(1);await connect.close();process.exit(0);
})();function sleep(time: number) {return new Promise((resolve) => setTimeout(resolve, time*1000));
}function random(max: number){return Math.floor(Math.random() * Math.floor(max));
}

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

// receive_logs_topic.ts
import * as amqp from 'amqplib'
const url = `amqp://localhost`;(async function receive(){const connect = await amqp.connect(url);const channel = await connect.createChannel();const exchange = 'topic_logs';await channel.assertExchange(exchange, 'topic', { durable: false });const queueA = await channel.assertQueue('queue_topic_A');// # 匹配多个单词await channel.bindQueue(queueA.queue, exchange, '*.orange.*');const queueB = await channel.assertQueue('queue_topic_B');// * 可以替代一个单词await channel.bindQueue(queueB.queue, exchange, '#');channel.consume(queueA.queue, msg => {console.log("队列AAAAA：", msg.content.toString())}, { noAck: true });channel.consume(queueB.queue, msg => {console.log("队列BBBBBBBB：", msg.content.toString())}, { noAck: true });// await connect.close();
})();

1
2
3
4
5
6
7
8
9
10

# 执行
$ ts-node emit_log_topic.ts
[x] Sent hello world-- error
[x] Sent hello world-- info$ ts-node receive_logs_topic.ts
队列AAAAA： hello world
队列BBBBBBBB： hello world队列AAAAA： hello world